Flink CDC 同步 Mysql 数据

一、Flink CDC、Flink、CDC各有啥关系

  Flink：流式盘算框架，不包含 Flink CDC，和 Flink CDC不要紧
  CDC：是一种思想，理念，不涉及某一门具体的技术。CDC 是变更数据捕捉（Change Data Capture）技术的缩写，它可以将源数据库（Source）的增量变更记录，同步到一个或多个数据目的（Sink）。在同步过程中，还可以对数据进行一定的处理，例如过滤、关联、分组、统计等。目前专业做数据库事件担当和剖析的中间件是Debezium，如果是捕捉Mysql，还有Canal。
  Flink CDC：是 CDC 的一种实现而已，不属于 Flink 子版块。这个技术是阿里开辟的。目的是为了丰富 Flink 的生态。
1.1 概述

  Flink CDC 基于数据库日志的 Change Data Caputre 技术，实现了全量和增量的一体化读取能力，并借助 Flink 良好的管道能力和丰富的上下游生态，支持捕捉多种数据库的变更，并将这些变更实时同步到下游存储。
1.2 和 jdbc Connectors 对比

  JDBC Connectors 连接器，确实可以读取外部的 数据库。比如：MySQL、Oracle、SqlServer等。但是，JDBC连数据库，只是瞬时操纵，没办法连续监听数据库的数据变化。
  Flink CDC Connectors，可以实现数据库的变更捕捉，能够连续不断地把变更数据同步到下游的体系中。
官网概述：https://ververica.github.io/flink-cdc-connectors/
github链接：https://github.com/ververica/flink-cdc-connectors
二、利用

  FlinkCDC 同步数据，有两种方式，一种是 FlinkSQL 的方式，一种是Flink DataStream 和 Table API 的方式。
  我这里直接用的是 ieda 测试的 DataStream 方式。
  代码来自：https://github.com/yclxiao/flink-cdc-demo/tree/main/src/main/java/com/yclxiao/flinkcdcdemo
  CloudAcctProfit2DwsHdjProfitRecordAPI.java

1. import com.alibaba.fastjson.JSON;
2. import com.alibaba.fastjson.JSONObject;
3. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
4. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
5. import com.xiaoqiang.utils.JdbcUtil;
6. import org.apache.commons.lang3.StringUtils;
7. import org.apache.commons.lang3.time.DateFormatUtils;
8. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
9. import org.apache.flink.api.common.functions.FilterFunction;
10. import org.apache.flink.api.common.functions.FlatMapFunction;
11. import org.apache.flink.configuration.Configuration;
12. import org.apache.flink.streaming.api.datastream.DataStreamSource;
13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
15. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
16. import org.apache.flink.util.Collector;
17. import org.slf4j.Logger;
18. import org.slf4j.LoggerFactory;
20. import java.sql.Connection;
21. import java.sql.DriverManager;
22. import java.sql.ResultSet;
23. import java.sql.Statement;
24. import java.util.*;
26. public class CloudAcctProfit2DwsHdjProfitRecordAPI {
27.     private static final Logger LOG = LoggerFactory.getLogger(CloudAcctProfit2DwsHdjProfitRecordAPI.class);
28.     private static String MYSQL_HOST = "x.x.x.x";
29.     private static int MYSQL_PORT = 3306;
30.     private static String MYSQL_USER = "root";
31.     private static String MYSQL_PASSWD = "xiaoqiang";
32.     private static String SYNC_DB = "league_test";
33.     private static List<String> SYNC_TABLES = Arrays.asList("league_test.oc_settle_profit");
35.     public static void main(String[] args) throws Exception {
36.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
37.                 .hostname(MYSQL_HOST)
38.                 .port(MYSQL_PORT)
39.                 .databaseList(SYNC_DB) // set captured database
40.                 .tableList(String.join(",", SYNC_TABLES)) // set captured table
41.                 .username(MYSQL_USER)
42.                 .password(MYSQL_PASSWD)
43.                 .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
44.                 .build();
46.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
47.         env.setParallelism(1);
48.         env.enableCheckpointing(5000);
50.         DataStreamSource<String> cdcSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "CDC Source" + "xiaoqiang-flink");
52.         List<String> tableList = getTableList();
53.         System.out.println("tableList--->"+tableList);
54.         for (String tbl : tableList) {
55.             SingleOutputStreamOperator<String> filterStream = filterTableData(cdcSource, "oc_settle_profit");
56. //            SingleOutputStreamOperator<String> cleanStream = clean(filterStream);
57.             // 流的数据sink出去
58.             filterStream.addSink(new CustomDealDataSink())
59.                     .name("sink " + tbl);
60.         }
61.         env.execute("xiaoqiang-flink");
62.     }
64.     /**
65.      * 自定义sink
66.      */
67.     private static class CustomDealDataSink extends RichSinkFunction<String> {
68.         private transient Connection coalitiondbConnection;
69.         private transient Statement coalitiondbStatement;
70.         private transient Connection cloudConnection;
71.         private transient Statement cloudStatement;
73.         @Override
74.         public void open(Configuration parameters) throws Exception {
75.             super.open(parameters);
76.             // 在这里初始化 JDBC 连接
77.             coalitiondbConnection = DriverManager.getConnection("jdbc:mysql://x.x.x.x:3306/league_test", "root", "");
78.             coalitiondbStatement = coalitiondbConnection.createStatement();
79.             cloudConnection = DriverManager.getConnection("jdbc:mysql://x.x.x.x:3306/cloud_test", "root", "");
80.             cloudStatement = cloudConnection.createStatement();
81.         }
83.         @Override
84.         public void invoke(String value, Context context) throws Exception {
85.             // 解析拿到的CDC-JSON数据
86.             JSONObject rowJson = JSON.parseObject(value);
87.             String outNo = rowJson.getString("out_no");
88.             Integer userType = rowJson.getInteger("user_type");
89.             String id = rowJson.getString("id");
90.             String payOrderNo = rowJson.getString("pay_order_no");
91.             String title = rowJson.getString("title");
92.             String fromUserId = rowJson.getString("from_user_id");
93.             String fromAccountId = rowJson.getString("from_account_id");
94.             String userId = rowJson.getString("user_id");
95.             String accountId = rowJson.getString("account_id");
96.             Integer amount = rowJson.getInteger("amount");
97.             Integer profitState = rowJson.getInteger("profit_state");
98.             Date profitTime = rowJson.getTimestamp("profit_time");
99.             Integer refundState = rowJson.getInteger("refund_state");
100.             Date refundTime = rowJson.getTimestamp("refund_time");
101.             Date addTime = rowJson.getTimestamp("add_time");
102.             String remark = rowJson.getString("remark");
103.             String acctCircle = rowJson.getString("acct_circle");
104.             Integer fromUserType = rowJson.getInteger("from_user_type");
105.             String companyId = rowJson.getString("company_id");
106.             String bizCompanyId = rowJson.getString("biz_company_id");
107. //            if (1 != profitState || !"PG11111".equals(acctCircle)) {
108. //                return;
109. //            }
110. //
111. //            // 读取相关表的数据（与其他表进行关联）
112. //            Integer bizType = null;
113. //            String contributeUserId = null;
114. //            String relationBrandOwnerId = null;
115. //            ResultSet virtualOrderResultSet = coalitiondbStatement.executeQuery("select * from tc_virtual_order where order_type != 2 and id = '" + outNo + "'");
116. //            // 如果是tc_virtual_order订单（上岗卡、安心卡、课程）
117. //            if (virtualOrderResultSet.next()) {
118. //                // 处理数据逻辑
119. //                Integer virtualOrder4OrderType = virtualOrderResultSet.getInt("order_type");
120. //                String virtualOrder4CompanyId = virtualOrderResultSet.getString("company_id");
121. //                String virtualOrder4BrandId = virtualOrderResultSet.getString("brand_id");
122. //                // 上岗卡订单排掉，因为已经有别的任务处理了
123. //                if (virtualOrder4OrderType == 2) {
124. //                    return;
125. //                }
126. //                // orderType转换
127. //                if (virtualOrder4OrderType == 6) {
128. //                    bizType = 10;
129. //                } else if (virtualOrder4OrderType == 1) {
130. //                    bizType = 11;
131. //                } else if (virtualOrder4OrderType == 5) {
132. //                    bizType = 12;
133. //                }
134. //                // userType转换
135. //                if (virtualOrder4OrderType == 6 && userType == 92) {
136. //                    contributeUserId = virtualOrder4CompanyId;
137. //                } else if (virtualOrder4OrderType == 1 && userType == 92) {
138. //                    contributeUserId = virtualOrder4CompanyId;
139. //                } else if (virtualOrder4OrderType == 5 && userType == 92) {
140. //                    contributeUserId = virtualOrder4CompanyId;
141. //                }
142. //                // relationBrandOwnerId转换
143. //                if (virtualOrder4OrderType == 6 && userType == 90) {
144. //                    relationBrandOwnerId = virtualOrder4BrandId;
145. //                } else if (virtualOrder4OrderType == 1 && userType == 90) {
146. //                    relationBrandOwnerId = virtualOrder4BrandId;
147. //                } else if (virtualOrder4OrderType == 5 && userType == 90) {
148. //                    relationBrandOwnerId = virtualOrder4BrandId;
149. //                }
150. //                // remark转换
151. //                if (virtualOrder4OrderType == 1 || virtualOrder4OrderType == 5) {
152. //                    remark = title;
153. //                }
154. //            } else {
155. //                // 如果不是tc_virtual_order的数据，则可能是其他数据，此处只保留好到家实物商品数据
156. //                if (StringUtils.isBlank(payOrderNo)) {
157. //                    return;
158. //                }
159. //                ResultSet acctPayOrderResultSet = cloudStatement.executeQuery("select * from acct_pay_order t where t.id = '" + payOrderNo + "'");
160. //                if (!acctPayOrderResultSet.next()) {
161. //                    return;
162. //                }
163. //                Integer payCate = acctPayOrderResultSet.getInt("pay_cate");
164. //                if (200100 != payCate) { // 好到家实物商品类型
165. //                    return;
166. //                }
167. //
168. //                bizType = 20;
169. //                if (userType == 92 && StringUtils.isNotBlank(bizCompanyId)) {
170. //                    contributeUserId = bizCompanyId;
171. //                } else if (userType == 90 && StringUtils.isNotBlank(bizCompanyId)) {
172. //                    ResultSet brandOwnerIdResultSet = cloudStatement.executeQuery("select * from uc_brand_partner t where t.company_id = '" + bizCompanyId + "'");
173. //                    if (brandOwnerIdResultSet.next()) {
174. //                        relationBrandOwnerId = brandOwnerIdResultSet.getString("brand_owner_id");
175. //                    }
176. //                }
177. //            }
178. //            if (StringUtils.isBlank(remark)) {
179. //                remark = title;
180. //            }
182.             // 数据写入到mysql
183.             String insertSql = "INSERT INTO dws_profit_record_hdj_flink_api (id, show_profit_id, order_no, from_user_id, from_user_type, user_id,\n" +
184.                     "                                                    user_type, amount, profit_time, state, acct_circle, biz_type,\n" +
185.                     "                                                    contribute_user_id, relation_brand_owner_id, remark, add_time)\n" +
186.                     "VALUES ('" + id + "', '" + "JSD" + id + "', '" + outNo + "', '" + fromUserId + "', " + fromUserType + ", '" + userId + "', " + userType + ",\n" +
187.                     "        " + amount + ", '" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss", TimeZone.getTimeZone("GMT")) + "', " + profitState + ", '" + acctCircle + "', " + 1 + ", " + (StringUtils.isBlank("123") ? null : "'" + "contributeUserId" + "'") + ", " + (StringUtils.isBlank("relationBrandOwnerId") ? null : "'" + "relationBrandOwnerId" + "'") + ", '" + remark + "',\n" +
188.                     "        '" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss", TimeZone.getTimeZone("GMT")) + "');";
189.             cloudStatement.execute("delete from dws_profit_record_hdj_flink_api where id = '" + id + "'");
190.             System.out.println("insertSql--->"+insertSql);
191.             cloudStatement.execute(insertSql);
192.         }
194.         @Override
195.         public void close() throws Exception {
196.             super.close();
197.             // 在这里关闭 JDBC 连接
198.             coalitiondbStatement.close();
199.             coalitiondbConnection.close();
200.             cloudStatement.close();
201.             cloudConnection.close();
202.         }
203.     }
205.     /**
206.      * 清晰数据
207.      *
208.      * @param source
209.      * @return
210.      */
211.     private static SingleOutputStreamOperator<String> clean(SingleOutputStreamOperator<String> source) {
212.         return source.flatMap(new FlatMapFunction<String, String>() {
213.             @Override
214.             public void flatMap(String row, Collector<String> out) throws Exception {
215.                 try {
216.                     LOG.info("============================row:{}", row);
217.                     JSONObject rowJson = JSON.parseObject(row);
218.                     String op = rowJson.getString("op");
219.                     //history,insert,update
220.                     if (Arrays.asList("r", "c", "u").contains(op)) {
221.                         out.collect(rowJson.getJSONObject("after").toJSONString());
222.                     } else {
223.                         LOG.info("filter other op:{}", op);
224.                     }
225.                 } catch (Exception ex) {
226.                     LOG.warn("filter other format binlog:{}", row);
227.                 }
228.             }
229.         });
230.     }
232.     /**
233.      * 过滤数据
234.      *
235.      * @param source
236.      * @param table
237.      * @return
238.      */
239.     private static SingleOutputStreamOperator<String> filterTableData(DataStreamSource<String> source, String table) {
240.         return source.filter(new FilterFunction<String>() {
241.             @Override
242.             public boolean filter(String row) throws Exception {
243.                 try {
244.                     JSONObject rowJson = JSON.parseObject(row);
245.                     JSONObject source = rowJson.getJSONObject("source");
246.                     String tbl = source.getString("table");
247.                     return table.equals(tbl);
248.                 } catch (Exception ex) {
249.                     ex.printStackTrace();
250.                     return false;
251.                 }
252.             }
253.         });
254.     }
256.     private static List<String> getTableList() {
257.         List<String> tables = new ArrayList<>();
258.         String sql = "SELECT TABLE_SCHEMA,TABLE_NAME FROM information_schema.tables WHERE TABLE_SCHEMA = '" + SYNC_DB + "'";
259.         List<JSONObject> tableList = JdbcUtil.executeQuery(MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWD, sql);
260.         for (JSONObject jsob : tableList) {
261.             String schemaName = jsob.getString("TABLE_SCHEMA");
262.             String tblName = jsob.getString("TABLE_NAME");
263.             String schemaTbl = schemaName + "." + tblName;
264.             if (SYNC_TABLES.contains(schemaTbl)) {
265.                 tables.add(tblName);
266.             }
267.         }
268.         return tables;
269.     }
270. }

复制代码

  JdbcUtil.java

1. import com.alibaba.fastjson.JSONObject;
2. import org.slf4j.Logger;
3. import org.slf4j.LoggerFactory;
5. import java.sql.*;
6. import java.util.ArrayList;
7. import java.util.List;
9. public class JdbcUtil {
11.     static {
12.         try {
13. //            Class.forName("com.mysql.cj.jdbc.Driver");
14.             Class.forName("com.mysql.jdbc.Driver");
15.         } catch (ClassNotFoundException e) {
16.             e.printStackTrace();
17.         }
18.     }
20.     private static final Logger LOG = LoggerFactory.getLogger(JdbcUtil.class);
22.     public static void main(String[] args) throws SQLException {
23.     }
25.     public static List<JSONObject> executeQuery(String hostUrl, int port, String user, String password, String sql) {
26.         List<JSONObject> beJson = new ArrayList<>();
27.         String connectionUrl = String.format("jdbc:mysql://%s:%s/league_test?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai", hostUrl, port);
28.         Connection con = null;
29.         try {
30.             con = DriverManager.getConnection(connectionUrl, user, password);
31.             PreparedStatement ps = con.prepareStatement(sql);
32.             ResultSet rs = ps.executeQuery();
33.             beJson = resultSetToJson(rs);
34.         } catch (SQLException e) {
35.             e.printStackTrace();
36.         } catch (Exception e) {
37.             e.printStackTrace();
38.         } finally {
39.             try {
40.                 con.close();
41.             } catch (Exception e) {
42.             }
43.         }
44.         return beJson;
45.     }
47.     private static List<JSONObject> resultSetToJson(ResultSet rs) throws SQLException {
48.         List<JSONObject> list = new ArrayList<>();
49.         ResultSetMetaData metaData = rs.getMetaData();
50.         int columnCount = metaData.getColumnCount();
51.         while (rs.next()) {
52.             JSONObject jsonObj = new JSONObject();
53.             for (int i = 1; i <= columnCount; i++) {
54.                 String columnName = metaData.getColumnLabel(i);
55.                 String value = rs.getString(columnName);
56.                 jsonObj.put(columnName, value);
57.             }
58.             list.add(jsonObj);
59.         }
60.         return list;
61.     }
62. }

复制代码

  pom.xml：

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

复制代码

2.1 Mysql 打开 bin-log 功能

  og_bin 的Value如果为ON代表开启，如果为OFF代表关闭，MySQL8.0默认是开启的：

1. # 查看是否开启binlog
2. mysql> SHOW VARIABLES LIKE '%log_bin%';

复制代码

  关闭状态：

• log_bin为ON代表MySQL已经开启binlog日志记录
  
• log_bin_basename设置了binlog的文件路径及文件前缀名
  
• log_bin_index设置了binlog索引文件的路径
  

  开启状态：

1. # 在centos中mysql的配置文件一般都在/etc/mysql目录下，如果不在可以通过 find / -name "my.cnf" 查找
2. vi /etc/mysql/my.cnf
4. # 服务ID
5. server-id=1
6. # binlog 配置 只要配置了log_bin地址 就会开启
7. log_bin = /var/lib/mysql/mysql_bin
8. # 日志存储天数 默认0 永久保存
9. # 如果数据库会定期归档，建议设置一个存储时间不需要一直存储binlog日志，理论上只需要存储归档之后的日志
10. expire_logs_days = 30
11. # binlog最大值
12. max_binlog_size = 1024M
13. # 规定binlog的格式，binlog有三种格式statement、row、mixad，默认使用statement，建议使用row格式
14. binlog_format = ROW
15. # 在提交n次事务后，进行binlog的落盘，0为不进行强行的刷新操作，而是由文件系统控制刷新日志文件，如果是在线交易和账有关的数据建议设置成1，如果是其他数据可以保持为0即可
16. sync_binlog = 1
18. # 重启MySQL服务使配置生效
19. systemctl restart mysqld / service mysql restart
21. # 查看日志列表
22. SHOW MASTER LOGS;

复制代码

可参考：MySQL 开启设置binlog以及通过binlog恢复数据
2.2 在 Mysql 中建库建表准备

1. CREATE DATABASE IF NOT EXISTS cloud_test;
2. CREATE DATABASE IF NOT EXISTS league_test;
4. CREATE TABLE league_test.oc_settle_profit (
5.         id                           varchar(32),
6.         show_profit_id               varchar(32),
7.         order_no                     varchar(32),
8.         from_user_id                 varchar(32),
9.         from_user_type               int(11),
10.         user_id                      varchar(32),
11.         user_type                    int(11),
12.         rate                         int(11),
13.         amount                       int(11),
14.         type                         int(11),
15.         add_time                     datetime,
16.         state                        int(11),
17.         expect_profit_time           datetime,
18.         profit_time                  datetime,
19.         profit_mode                  int(11),
20.         opt_code                     varchar(32),
21.         opt_name                     varchar(32),
22.         acct_circle                  varchar(32),
23.         process_state                int(11),
24.         parent_id                    varchar(32),
25.         keep_account_from_user_id    varchar(32),
26.         keep_account_from_bm_user_id varchar(32),
27.         keep_account_user_id         varchar(32),
28.         keep_account_bm_user_id      varchar(32),
29.         biz_type                     int(11),
30.         remark                       varchar(32),
31.         contribute_user_id           varchar(32),
32.         relation_brand_owner_id      varchar(32),
33.         PRIMARY KEY (id) USING BTREE
34. );
36. CREATE TABLE cloud_test.dws_profit_record_hdj_flink_api (
37.         id                      varchar(32),
38.         show_profit_id          varchar(32),
39.         order_no                varchar(32),
40.         from_user_id            varchar(32),
41.         from_user_type          int(11),
42.         user_id                 varchar(32),
43.         user_type               int(11),
44.         amount                  int(11),
45.         profit_time             datetime,
46.         state                   int(11),
47.         acct_circle             varchar(32),
48.         biz_type                int(11),
49.         contribute_user_id      varchar(32),
50.         relation_brand_owner_id varchar(32),
51.         remark                  varchar(32),
52.         add_time                datetime,
53.         PRIMARY KEY (id) USING BTREE
54.         );

复制代码

2.3 遇到的坑

2.3.1 The MySQL server has a timezone offset (0 seconds ahead of UTC)

  用 JDBC 连接 Mysql 的时间报错：The MySQL server has a timezone offset (0 seconds ahead of UTC)
  缘故原由：从错误即可知道是时区的错误。

1. show variables like '%time_zone%';
2. Variable_name   |Value |
3. ----------------+------+
4. time_zone       |SYSTEM|
6. // 或者下面这条命令
7. SELECT @@global.time_zone;

复制代码

  办理：利用 root 用户登录 mysql，再执行 set global time_zone='+8:00' 命令。
  注意：一开始改成了 SET GLOBAL time_zone = 'Asia/Shanghai'，但并不好使。
2.3.2 自动转换datetime为时间戳题目

  Apache Flink 是一种流处理框架，它可以读取 MySQL binlog 日志，并将其转换为流数据。然而，由于Flink内部采取的时间戳格式与 MySQL 的 datetime 格式差别，所以在抓取 binlog 时，须要进行一定的转换才华正确地剖析数据。
  自界说类：DateTimeConverter 实现 CustomConverter 接口，重写对应方法对 mysql 的时间类型进行尺度转换

1. import io.debezium.spi.converter.CustomConverter;
2. import io.debezium.spi.converter.RelationalColumn;
3. import org.apache.kafka.connect.data.SchemaBuilder;
5. import java.time.*;
6. import java.time.format.DateTimeFormatter;
7. import java.util.Properties;
8. /**
9. * @BelongsProject:
10. * @BelongsPackage:
11. * @Author:
12. * @CreateTime:
13. * @Description: TODO 实现CustomConverter接口，重写对应方法对mysql的时间类型进行标准转换
14. * @Version: 1.0
15. */
16. public class DateTimeConverter implements CustomConverter<SchemaBuilder, RelationalColumn>{
17.     private DateTimeFormatter dateFormatter = DateTimeFormatter.ISO_DATE;
19.     private DateTimeFormatter timeFormatter = DateTimeFormatter.ISO_TIME;
21.     private DateTimeFormatter datetimeFormatter = DateTimeFormatter.ISO_DATE_TIME;
23.     private DateTimeFormatter timestampFormatter = DateTimeFormatter.ISO_DATE_TIME;
25.     private ZoneId timestampZoneId = ZoneId.systemDefault();
27.     @Override
28.     public void configure(Properties props) {
30.     }
32.     @Override
33.     public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
34.         String sqlType = column.typeName().toUpperCase();
35.         SchemaBuilder schemaBuilder = null;
36.         Converter converter = null;
38.         if ("DATE".equals(sqlType)) {
39.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.date.string");
40.             converter = this::convertDate;
41.         }
43.         if ("TIME".equals(sqlType)) {
44.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.time.string");
45.             converter = this::convertTime;
46.         }
48.         if ("DATETIME".equals(sqlType)) {
49.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.datetime.string");
50.             converter = this::convertDateTime;
51.         }
53.         if ("TIMESTAMP".equals(sqlType)) {
54.             schemaBuilder = SchemaBuilder.string().optional().name("com.darcytech.debezium.timestamp.string");
55.             converter = this::convertTimestamp;
56.         }
58.         if (schemaBuilder != null) {
59.             registration.register(schemaBuilder, converter);
60.         }
61.     }
62.    
63.     private String convertDate(Object input) {
64.         if (input == null) return null;
66.         if (input instanceof LocalDate) {
67.             return dateFormatter.format((LocalDate) input);
68.         }
69.         
70.         if (input instanceof Integer) {
71.             LocalDate date = LocalDate.ofEpochDay((Integer) input);
72.             return dateFormatter.format(date);
73.         }
74.         return String.valueOf(input);
75.     }
78.     private String convertTime(Object input) {
79.         if (input == null) return null;
81.         if (input instanceof Duration) {
82.             Duration duration = (Duration) input;
83.             long seconds = duration.getSeconds();
84.             int nano = duration.getNano();
86.             LocalTime time = LocalTime.ofSecondOfDay(seconds).withNano(nano);
87.             return timeFormatter.format(time);
88.         }
89.         return String.valueOf(input);
90.     }
93.     private String convertDateTime(Object input) {
94.         if (input == null) return null;
96.         if (input instanceof LocalDateTime) {
97.             return datetimeFormatter.format((LocalDateTime) input).replaceAll("T", " ");
98.         }
99.         return String.valueOf(input);
100.     }
103.     private String convertTimestamp(Object input) {
104.         if (input == null) return null;
106.         if (input instanceof ZonedDateTime) {
107.             // mysql的timestamp会转成UTC存储，这里的zonedDatetime都是UTC时间
108.             ZonedDateTime zonedDateTime = (ZonedDateTime) input;
109.             LocalDateTime localDateTime = zonedDateTime.withZoneSameInstant(timestampZoneId).toLocalDateTime();
110.             return timestampFormatter.format(localDateTime).replaceAll("T", " ");
111.         }
112.         return String.valueOf(input);
113.     }
114. }

复制代码

  引入：

1. public class FlinkSourceUtil {
3.     public static MySqlSource<JSONObject> getMySqlSource(ParameterTool parameterTool,List<String> databases, List<String> tables){
4.         Properties props = new Properties();
5.         props.setProperty("useSSL", "false");
6.         props.setProperty("allowPublicKeyRetrieval", "true");
8.         Properties debeziumProperties = new Properties();
9.         debeziumProperties.setProperty("converters", "dateConverters");
10.         debeziumProperties.setProperty("dateConverters.type", "com.xxx.util.DateTimeConverter");
11.         debeziumProperties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
12.         debeziumProperties.setProperty("dateConverters.format.time", "HH:mm:ss");
13.         debeziumProperties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
14.         debeziumProperties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
15.         debeziumProperties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");
16.         
17.         String[] databaseArray = databases.toArray(new String[0]);
18.         String[] tableArray = tables.toArray(new String[0]);
20.         MySqlSource<JSONObject> mySqlSource = MySqlSource.<JSONObject>builder()
21.                 .hostname(parameterTool.get(Constant.MYSQl_SOURCE_HOST_NAME))
22.                 .port(Integer.parseInt(parameterTool.get(Constant.MYSQl_SOURCE_PORT)))
23.                 .databaseList(databaseArray)
24.                 .tableList(tableArray)
25.                 .username(parameterTool.get(Constant.MYSQL_SOURCE_USER_NAME))
26.                 .password(parameterTool.get(Constant.MYSQL_SOURCE_PASSWORD))
27.                 .deserializer(new MyDebeziumDeserializationSchema())
28.                 .debeziumProperties(debeziumProperties)
29.                 .startupOptions(StartupOptions.initial())
30.                 .serverId(parameterTool.get(Constant.SERVER_ID))
31.                 .jdbcProperties(props)
32.                 .build();
33.         return mySqlSource;
34.     }
35. }

复制代码

参考：
关于flinkCDC监听MySQL binlog时自动转换datetime为时间戳题目
FlinkCDC时间题目timestamp等
flink-core抓mysql-binlog，字段datetime会自动转换成时间戳，怎么办理？
2.3.3 重起程序会全量读取 binlog，我想要的是增量

  其实 MySqlSourceBuilder 是有一个方法特意指定 startUP mode 的，改造代码

1.         MySqlSourceBuilder<String> mySqlSource = new MySqlSourceBuilder<>();
2.         mySqlSource.startupOptions(StartupOptions.latest());
4.         mySqlSource
5.                 .hostname(MYSQL_HOST)
6.                 .port(MYSQL_PORT)
7.                 .databaseList(SYNC_DB) // set captured database
8.                 .tableList(SYNC_TABLES) // set captured table
9.                 .username(MYSQL_USER)
10.                 .password(MYSQL_PASSWD)
11.                 .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
12.                 .debeziumProperties(debeziumProperties)
13.                 .build();
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.         env.setParallelism(1);
17.         env.enableCheckpointing(5000);
19.         DataStreamSource<String> cdcSource = env.fromSource(mySqlSource.build(), WatermarkStrategy.noWatermarks(), "CDC Source" + "flinkcdc-kafka");

复制代码

参考：Flink cdc怎样只进行增量同步，差别步汗青数据（只读取binlog)
2.4 测试

  Idea 启动程序后，在 oc_settle_profit 表中插入数据后 dws_profit_record_hdj_flink_api 也可以同步插入相应的数据。
参考：
【博学谷学习记录】超强总结，专心分享|大数据之flinkCDC
一次打通FlinkCDC同步Mysql数据
三、番外

  用 Flink CDC 可以监控 Mysql，但无法监控 StarRocks，和官方扣问过，目前 StarRocks 并没有像 Mysql 这样被外部感知 DDL 操纵的 bin-log 功能，所以暂时还无法用 Flink CDC 监控 StarRocks。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。