devbox
Cpp五条
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink CDC 将MySQL的数据写入Hudi实践_scan.incremental.snapshot.enabled

作者：Cpp五条 | 2024-05-20 15:48:07

赞

踩

scan.incremental.snapshot.enabled

Flink CDC + Hudi实践

一、依赖关系

1、Maven依赖

2、SQL客户端JAR

二、设置MySQL服务器

1、创建MySQL用户：

2、向用户授予所需的权限：

3、最终确定用户的权限：

三、注意

1、MySQL CDC源代码如何工作

2、向MySQL用户授予RELOAD权限

3、全局读取锁（FLUSH TABLES WITH READ LOCK）

4、为每个作业设置一个differnet SERVER ID

5、扫描数据库表期间无法执行检查点

6、设置MySQL会话超时

四.Flink cdc 数据入湖(Hudi)

1.创建mysql源表

2.使用Flink cdc mysql连接器创建flinkSQL映射表

3.创建FlinkSQL Hudi连接器创建hudi表

4.将数据从CDC表插入hudi表

5.直接用hudi表进行join操作

6.FlinkCDC会根据字段名映射mysql的源表,字段可以不一一对应.

一、依赖关系

        为了设置MySQL CDC连接器，下表提供了使用构建自动化工具（例如Maven或SBT）和带有SQL JAR捆绑包的SQL Client的两个项目的依赖项信息。

1、Maven依赖

  1. <dependency>
  2.   <groupId>com.ververica</groupId>
  3.   <artifactId>flink-connector-mysql-cdc</artifactId>
  4.   <version>2.0.0</version>
  5. </dependency>

2、SQL客户端JAR

        下载flink-sql-connector-mysql-cdc-2.0.0.jar并将其放在下<FLINK_HOME>/lib/。

二、设置MySQL服务器

        您必须定义一个对Debezium MySQL连接器监视的所有数据库具有适当权限的MySQL用户。

1、创建MySQL用户：

mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';

2、向用户授予所需的权限：

mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

3、最终确定用户的权限：

mysql> FLUSH PRIVILEGES;

查看有关权限说明的更多信息： https://debezium.io/documentation/reference/1.2/connectors/mysql.html#_permissions_explained

1、MySQL CDC源代码如何工作

        启动MySQL CDC源时，它将获取一个全局读取锁（FLUSH TABLES WITH READ LOCK），该锁将阻止其他数据库的写入。然后，它读取当前binlog位置以及数据库和表的schema。之后，将释放 全局读取锁。然后，它扫描数据库表并从先前记录的位置读取binlog。Flink将定期执行checkpoints以记录binlog位置。如果发生故障，作业将重新启动并从checkpoint完成的binlog位置恢复。因此，它保证了仅一次的语义。

2、向MySQL用户授予RELOAD权限

        如果未授予MySQL用户RELOAD权限，则MySQL CDC源将改为使用表级锁，并使用此方法执行快照。这会阻止写入更长的时间。

3、全局读取锁（FLUSH TABLES WITH READ LOCK）

        全局读取锁 在读取binlog位置和schema期间保持。这可能需要几秒钟，具体取决于表的数量。全局读取锁定会阻止写入，因此它仍然可能影响在线业务。

如果要跳过读取锁，并且可以容忍至少一次语义，则可以添加'debezium.snapshot.locking.mode' = 'none'选项以跳过锁。

4、为每个作业设置一个differnet SERVER ID

每个用于读取binlog的MySQL数据库客户端都应具有唯一的ID，称为server id。MySQL服务器将使用此ID维护网络连接和binlog位置。如果不同的作业共享相同的server id，则可能导致从错误的binlog位置进行读取。

提示：默认情况下，启动TaskManager时，server id是随机的。如果TaskManager失败，则再次启动时，它可能具有不同的server id。但这不应该经常发生（作业异常不会重新启动TaskManager），也不会对MySQL服务器造成太大影响。

因此，建议为每个作业设置不同的server id ，例如：

  • 通过SQL Hints： SELECT * FROM source_table /+ OPTIONS('server-id'='123456') / ;
  • 通过Stream ApI的 创建source时设置： ** MySQLSource.builder().xxxxxx.serverId(123456);**

重点：Mysq的binlog 可以说是针对库级别，所以相同的server id去拉一个库里的不同表或者相同表可能会造成数据丢失。所以建议设置server id。（我也有在社区邮件中提问Jark大佬：传送地址）

5、扫描数据库表期间无法执行检查点

        在扫描表期间，由于没有可恢复的位置，因此我们无法执行checkpoints。为了不执行检查点，MySQL CDC源将保持检查点等待超时。超时检查点将被识别为失败的检查点，默认情况下，这将触发Flink作业的故障转移。因此，如果数据库表很大，则建议添加以下Flink配置，以避免由于超时检查点而导致故障转移：

  1. execution.checkpointing.interval: 10min
  2. execution.checkpointing.tolerable-failed-checkpoints: 100
  3. restart-strategy: fixed-delay
  4. restart-strategy.fixed-delay.attempts: 2147483647

 6、设置MySQL会话超时

        为大型数据库创建初始一致的快照时，在读取表时，您建立的连接可能会超时。您可以通过在MySQL配置文件中配置Interactive_timeout和wait_timeout来防止此行为。

  • interactive_timeout：服务器在关闭交互式连接之前等待活动的秒数。参见MySQL文档。
  • wait_timeout：服务器在关闭非交互式连接之前等待其活动的秒数。参见MySQL文档。

 四.Flink cdc 数据入湖(Hudi)

1.创建mysql源表

  1. -- MySQL
  2. /*Table structure for table `order_info` */
  3. DROP TABLE IF EXISTS `order_info`;
  4. CREATE TABLE `order_info` (
  5.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
  6.   `consignee` varchar(100) DEFAULT NULL COMMENT '收货人',
  7.   `consignee_tel` varchar(20) DEFAULT NULL COMMENT '收件人电话',
  8.   `total_amount` decimal(10,2) DEFAULT NULL COMMENT '总金额',
  9.   `order_status` varchar(20) DEFAULT NULL COMMENT '订单状态,1表示下单，2表示支付',
  10.   `user_id` bigint(20) DEFAULT NULL COMMENT '用户id',
  11.   `payment_way` varchar(20) DEFAULT NULL COMMENT '付款方式',
  12.   `delivery_address` varchar(1000) DEFAULT NULL COMMENT '送货地址',
  13.   `order_comment` varchar(200) DEFAULT NULL COMMENT '订单备注',
  14.   `out_trade_no` varchar(50) DEFAULT NULL COMMENT '订单交易编号（第三方支付用)',
  15.   `trade_body` varchar(200) DEFAULT NULL COMMENT '订单描述(第三方支付用)',
  16.   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
  17.   `operate_time` datetime DEFAULT NULL COMMENT '操作时间',
  18.   `expire_time` datetime DEFAULT NULL COMMENT '失效时间',
  19.   `tracking_no` varchar(100) DEFAULT NULL COMMENT '物流单编号',
  20.   `parent_order_id` bigint(20) DEFAULT NULL COMMENT '父订单编号',
  21.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片路径',
  22.   `province_id` int(20) DEFAULT NULL COMMENT '地区',
  23.   PRIMARY KEY (`id`)
  24. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单表';
  25. -- ----------------------------
  26. -- Records of order_info
  27. -- ----------------------------
  28. INSERT INTO `order_info` 
  29. VALUES (476, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9);
  30. INSERT INTO `order_info`
  31. VALUES (477, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3);
  32. INSERT INTO `order_info`
  33. VALUES (478, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7);
  34. INSERT INTO `order_info` 
  35. VALUES (480, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9);
  36. INSERT INTO `order_info`
  37. VALUES (481, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3);
  38. INSERT INTO `order_info`
  39. VALUES (482, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7);
  40. INSERT INTO `order_info` 
  41. VALUES (483, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9);
  42. INSERT INTO `order_info`
  43. VALUES (484, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3);
  44. INSERT INTO `order_info`
  45. VALUES (485, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7);
  46. INSERT INTO `order_info` 
  47. VALUES (486, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9);
  48. INSERT INTO `order_info`
  49. VALUES (487, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3);
  50. INSERT INTO `order_info`
  51. VALUES (488, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7);
  52. INSERT INTO `order_info` 
  53. VALUES (489, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9);
  54. INSERT INTO `order_info`
  55. VALUES (490, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3);
  56. INSERT INTO `order_info`
  57. VALUES (491, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7);
  58.  
  59. /*Table structure for table `order_detail` */
  60. CREATE TABLE `order_detail` (
  61.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
  62.   `order_id` bigint(20) DEFAULT NULL COMMENT '订单编号',
  63.   `sku_id` bigint(20) DEFAULT NULL COMMENT 'sku_id',
  64.   `sku_name` varchar(200) DEFAULT NULL COMMENT 'sku名称（冗余)',
  65.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片名称（冗余)',
  66.   `order_price` decimal(10,2) DEFAULT NULL COMMENT '购买价格(下单时sku价格）',
  67.   `sku_num` varchar(200) DEFAULT NULL COMMENT '购买个数',
  68.   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
  69.   PRIMARY KEY (`id`)
  70. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单明细表';
  71.  
  72. -- ----------------------------
  73. -- Records of order_detail
  74. -- ----------------------------
  75. INSERT INTO `order_detail` 
  76. VALUES (1329, 476, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://XLMByOyZDTJQYxphQHNTgYAFzJJCKTmCbzvEJIpz', 8900.00, '3', '2020-06-18 02:21:38');
  77. INSERT INTO `order_detail` 
  78. VALUES (1330, 477, 9, '荣耀10 GT游戏加速 AIS手持夜景 6GB+64GB 幻影蓝全网通 移动联通电信', 'http://ixOCtlYmlxEEgUfPLiLdjMftzrleOEIBKSjrhMne', 2452.00, '4', '2020-06-18 09:12:25');
  79. INSERT INTO `order_detail`
  80. VALUES (1331, 478, 4, '小米Play 流光渐变AI双摄 4GB+64GB 梦幻蓝 全网通4G 双卡双待 小水滴全面屏拍照游戏智能手机', 'http://RqfEFnAOqnqRnNZLFRvBuwXxwNBtptYJCILDKQYv', 1442.00, '1', '2020-06-18 15:56:34');
  81. INSERT INTO `order_detail` 
  82. VALUES (1332, 478, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://IwhuCDlsiLenfKjPzbJrIoxswdfofKhJLMzlJAKV', 8900.00, '3', '2020-06-18 15:56:34');
  83. INSERT INTO `order_detail` 
  84. VALUES (1333, 478, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://bbfwTbAzTWapywODzOtDJMJUEqNTeRTUQuCDkqXP', 8900.00, '1', '2020-06-18 15:56:34');

 2.使用Flink cdc mysql连接器创建flinkSQL映射表

  1. CREATE TABLE order_info(
  2.     id BIGINT NOT NULL PRIMARY KEY NOT ENFORCED,
  3.     user_id BIGINT,
  4.     create_time TIMESTAMP(0),
  5.     operate_time TIMESTAMP(0),
  6.     province_id INT,
  7.     order_status STRING,
  8.     total_amount DECIMAL(10, 5)
  9.   ) WITH (
  10.     'connector' = 'mysql-cdc',
  11.     'hostname' = '192.168.88.161',
  12.     'port' = '3306',
  13.     'username' = 'user',
  14.     'password' = 'password',
  15.     'database-name' = 'bigdata',
  16.     'table-name' = 'order_info',
  17.     'server-id' = '5401',
  18.     'scan.incremental.snapshot.enabled'='false'
  19. );
  20.  
  21. set sql-client.execution.result-mode=tableau;
  22.  
  23. select * from  order_info;
  24.  
  25. CREATE TABLE order_detail(
  26.     id BIGINT,
  27.     order_id BIGINT,
  28.     sku_id BIGINT,
  29.     sku_name STRING,
  30.     sku_num BIGINT,
  31.     order_price DECIMAL(10, 5),
  32. 	create_time TIMESTAMP(0)
  33.  ) WITH (
  34.     'connector' = 'mysql-cdc',
  35.     'hostname' = '192.168.88.161',
  36.     'port' = '3306',
  37.     'username' = 'user',
  38.     'password' = 'password',
  39.     'database-name' = 'bigdata',
  40.     'table-name' = 'order_detail',
  41.     'server-id' = '5402',
  42.     'scan.incremental.snapshot.enabled'='false'
  43. );
  44.  
  45. select * from  order_detail;

3.创建FlinkSQL Hudi连接器创建hudi表 

  1. CREATE TABLE order_info_hudi(
  2.     id BIGINT NOT NULL PRIMARY KEY NOT ENFORCED,
  3.     user_id BIGINT,
  4.     create_time TIMESTAMP(0),
  5.     operate_time TIMESTAMP(0),
  6.     province_id INT,
  7.     order_status STRING,
  8.     total_amount DECIMAL(10, 5)
  9.   ) WITH (
  10.    'connector' = 'hudi',
  11.    'path' = 'hdfs://node1:8020/hudi/order_info_hudi',
  12.    'table.type' = 'MERGE_ON_READ',
  13.    'changelog.enabled' = 'true',
  14.    'write.precombine.field' = 'create_time',
  15.    'compaction.async.enabled' = 'false'
  16. );
  17.  
  18.  
  19. CREATE TABLE order_detail_hudi(
  20.     id BIGINT,
  21.     order_id BIGINT,
  22.     sku_id BIGINT,
  23.     sku_name STRING,
  24.     sku_num BIGINT,
  25.     order_price DECIMAL(10, 5),
  26. 	create_time TIMESTAMP(0)
  27.  ) WITH (
  28.    'connector' = 'hudi',
  29.    'path' = 'hdfs://node1:8020/hudi/order_detail_hudi',
  30.    'table.type' = 'MERGE_ON_READ',
  31.    'changelog.enabled' = 'true',
  32.    'write.precombine.field' = 'create_time',
  33.    'hoodie.datasource.write.recordkey.field' = 'id',
  34.    'compaction.async.enabled' = 'false'
  35. );

4.将数据从CDC表插入hudi表

  1. insert into order_info_hudi select * from order_info;
  2.  
  3. select * from hoodie_table/*+ OPTIONS('read.streaming.enabled'='true')*/;
  4.  
  5. insert into order_detail_hudi select * from order_detail;
  6.  
  7. select * from order_detail;

 注意事项:

        同时可以查看HDFS里的Hudi数据路径，这里需要等Flink 5次checkpoint（默认配置可修改）之后才能查看到这些目录，一开始只有.hoodie一个文件夹

5.直接用hudi表进行join操作

  1. SELECT
  2.     od.id,
  3.     oi.id order_id,
  4.     oi.user_id,
  5.     oi.province_id,
  6.     od.sku_id,
  7.     od.sku_name,
  8.     od.sku_num,
  9.     od.order_price,
  10.     oi.create_time,
  11.     oi.operate_time
  12. FROM
  13.    (
  14.     SELECT * 
  15.     FROM order_info_hudi
  16.     WHERE 
  17. 	     order_status = '2'-- 已支付
  18.    ) oi
  19.    JOIN
  20.   (
  21.     SELECT *
  22.     FROM order_detail_hudi
  23.   ) od 
  24.   ON oi.id = od.order_id;

 6.FlinkCDC会根据字段名映射mysql的源表,字段可以不一一对应.

  1. CREATE TABLE order_info(
  2.     id BIGINT NOT NULL PRIMARY KEY NOT ENFORCED,
  3.     user_id BIGINT,
  4.     create_time TIMESTAMP(0),
  5.     total_amount DECIMAL(10, 5)
  6.   ) WITH (
  7.     'connector' = 'mysql-cdc',
  8.     'hostname' = '192.168.88.161',
  9.     'port' = '3306',
  10.     'username' = 'user',
  11.     'password' = 'password',
  12.     'database-name' = 'bigdata',
  13.     'table-name' = 'order_info',
  14.     'server-id' = '5401',
  15.     'scan.incremental.snapshot.enabled'='false'
  16. );

 查询的节结果是这样的

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/598293
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号