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

Flink CDC 2.0原理详解和生产实践

作者：花生_TL007 | 2024-05-17 02:56:28

赞

踩

com.ververica.cdc.debezium.debeziumdeserializationschema

点击上方蓝色字体，选择“设为星标”

回复"面试"获取更多惊喜

八股文交给我，你们专心刷题和面试

19fb146ec39c4b1250a5bc0099e299fa.png

Hi，我是王知无，一个大数据领域的原创作者。 

放心关注我，获取更多行业的一手消息。

Flink CDC 概念

CDC 的全称是 Change Data Capture ，在广义的概念上，只要能捕获数据变更的技术，我们都可以称为 CDC 。通常我们说的 CDC 技术主要面向 数据库的变更，是一种用于捕获数据库中数据变更的技术。

应用场景

  1. 数据同步，用于备份，容灾

  2. 数据分发，一个数据源分发给多个下游

  3. 数据采集(E)，面向数据仓库/数据湖的 ETL 数据集成

CDC 技术

目前业界主流的实现机制的可以分为两种：

  1. 基于查询的 CDC

  1. a.离线调度查询作业，批处理。
  2. b.无法保障数据一致性。
  3. c.不保障实时性。

  1. 基于日志的 CDC

  1. a.实时消费日志，流处理。
  2. b.保障数据一致性。
  3. c.提供实时数据。

常见的开源 CDC 方案

5f7d90e870be0050452a5267fe914209.png c9002e1fd87be8c413f7a4627293abe5.png 21f659c1787ab9b146f75c1117bb56f9.png 4f4811e23691ff4c48f3ce9307ed2d46.png cc9a77a4ec7aac63ef35005070329dd8.png 1d75beed858b14d2c3e42544aaf6ad84.png d351d8185ad77ac2296e3f9f3e84aa46.png d905b58e7945eb2ecc8ab4550686c611.png 5eaa815de0e009862fa60de830f05d7a.png

Flink CDC 2.0 设计详解

7199f22d31f7959ae3897e9b15067a56.png cc5f0ebf36df3853f905a3240fe3a59e.png 39e04bbe03ee675747e7e4e8dd98fc83.png 0ea8ee271aad894372125a0aa5cfec32.png 307d9a6732628e0bf67ad164a4572a31.png 5b48e501e23459e86f7b7b415f1b4609.png e434dbb235ab6cf0b690e5f05cbb1798.png 0945383da868e8d79a902c0f39ef583c.png 65a706dcabd86a603cf0e6257879a1c6.png c3280b7d22b81a0ed49c37423fa55fee.png c5957775f255208a76d753638ace799c.png

Source 官网 

https://github.com/ververica/flink-cdc-connectors

支持的连接

ab41e1b55dd8f4661c3778831916ac4b.png

实战应用

pom 文件

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>Flink-learning</artifactId>
  7.         <groupId>com.wudl.flink</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>Flink-cdc2.0</artifactId>
  13.     <properties>
  14.         <flink-version>1.13.0</flink-version>
  15.             <maven.compiler.source>1.8</maven.compiler.source>
  16.             <maven.compiler.target>1.8</maven.compiler.target>
  17.     </properties>
  18.  
  19.  
  20.  
  21.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.projectlombok</groupId>
  24.             <artifactId>lombok</artifactId>
  25.             <version>1.18.2</version>
  26.             <scope>provided</scope>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.apache.flink</groupId>
  30.             <artifactId>flink-java</artifactId>
  31.             <version>${flink-version}</version>
  32.         </dependency>
  33.  
  34.         <dependency>
  35.             <groupId>org.apache.flink</groupId>
  36.             <artifactId>flink-streaming-java_2.12</artifactId>
  37.             <version>${flink-version}</version>
  38.         </dependency>
  39.  
  40.         <dependency>
  41.             <groupId>org.apache.flink</groupId>
  42.             <artifactId>flink-clients_2.12</artifactId>
  43.             <version>${flink-version}</version>
  44.         </dependency>
  45.  
  46.         <dependency>
  47.             <groupId>org.apache.hadoop</groupId>
  48.             <artifactId>hadoop-client</artifactId>
  49.             <version>3.1.3</version>
  50.         </dependency>
  51.  
  52.         <dependency>
  53.             <groupId>mysql</groupId>
  54.             <artifactId>mysql-connector-java</artifactId>
  55.             <version>5.1.49</version>
  56.         </dependency>
  57.  
  58.         <dependency>
  59.             <groupId>org.apache.flink</groupId>
  60.             <artifactId>flink-table-planner-blink_2.12</artifactId>
  61.             <version>${flink-version}</version>
  62.         </dependency>
  63.  
  64.         <dependency>
  65.             <groupId>com.ververica</groupId>
  66.             <artifactId>flink-connector-mysql-cdc</artifactId>
  67.             <version>2.0.2</version>
  68.         </dependency>
  69.  
  70.         <dependency>
  71.             <groupId>com.alibaba</groupId>
  72.             <artifactId>fastjson</artifactId>
  73.             <version>1.2.75</version>
  74.         </dependency>
  75.  
  76.         <dependency>
  77.             <groupId>org.apache.flink</groupId>
  78.             <artifactId>flink-connector-jdbc_2.12</artifactId>
  79.             <version>1.13.3</version>
  80.         </dependency>
  81.     </dependencies>
  82.     <build>
  83.         <plugins>
  84.             <plugin>
  85.                 <groupId>org.apache.maven.plugins</groupId>
  86.                 <artifactId>maven-assembly-plugin</artifactId>
  87.                 <version>3.0.0</version>
  88.                 <configuration>
  89.                     <descriptorRefs>
  90.                         <descriptorRef>jar-with-dependencies</descriptorRef>
  91.                     </descriptorRefs>
  92.                 </configuration>
  93.                 <executions>
  94.                     <execution>
  95.                         <id>make-assembly</id>
  96.                         <phase>package</phase>
  97.                         <goals>
  98.                             <goal>single</goal>
  99.                         </goals>
  100.                     </execution>
  101.                 </executions>
  102.             </plugin>
  103.         </plugins>
  104.     </build>
  105. </project>

代码

  1. package com.wud.cdc2;
  2.  
  3. import com.ververica.cdc.connectors.mysql.MySqlSource;
  4. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  6. import com.ververica.cdc.debezium.DebeziumSourceFunction;
  7. import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  8. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  9. import org.apache.flink.streaming.api.CheckpointingMode;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.environment.StreamPipelineOptions;
  14. import org.apache.flink.table.api.Table;
  15. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  16. import org.apache.flink.types.Row;
  17.  
  18. import static org.apache.flink.table.api.Expressions.$;
  19.  
  20. public class FlinkCDC {
  21.     public static void main(String[] args) throws Exception {
  22.  
  23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  24.         env.setParallelism(1);
  25.         StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
  26.         env.enableCheckpointing(5000L);
  27.         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  28.         // 设置任务关闭时候保留最后一次checkpoint 的数据
  29.         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  30.         // 指定ck 的自动重启策略
  31.         env.setStateBackend(new FsStateBackend("hdfs://192.168.1.161:8020/cdc2.0-test/ck"));
  32.         // 设置hdfs 的访问用户名
  33.         System.setProperty("HADOOP_USER_NAME","hdfs");
  34.  
  35.         DebeziumSourceFunction<String> mySqlSource = MySqlSource.<String>builder()
  36.                 .hostname("192.168.1.180")
  37.                 .port(3306)
  38.                 .username("root")
  39.                 .password("123456")
  40.                 .databaseList("test")
  41.                 .tableList("test.Flink_iceberg")
  42.                 .deserializer(new StringDebeziumDeserializationSchema())
  43.                 .startupOptions(StartupOptions.initial())
  44.                 .build();
  45.         DataStreamSource<String> dataStreamSource = env.addSource(mySqlSource);
  46.         dataStreamSource.print();
  47.         env.execute();
  48.  
  49.  
  50.     }
  51. }

执行结果

  1. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585007,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585013}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  2. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585015,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585016}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  3. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  4. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  5. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  6. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  7. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10013,name=flink-mysqA3,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  8. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10014,name=flink-mysqA4,age=19,dt=2021-09-28},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  9. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585017}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  10. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585017,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  11. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  12. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  13. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10011,name=flink-mysql,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  14. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10012,name=flink-mysqA,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  15. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10013,name=flink-mysqA3,age=19,dt=2021-09-24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  16. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765, snapshot=true}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10014,name=flink-mysqA4,age=19,dt=2021-09-28},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=true,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  17. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1633178585, file=mysql-bin.000036, pos=765}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=10050,name=flink-cdc-add,age=21,dt=2021-10-2},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1633178585018,snapshot=last,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000036,pos=765,row=0},op=r,ts_ms=1633178585018}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}

集群提交

执行命令
  1. [root@basenode flink-1.13.2]# bin/flink run -c com.wud.cdc2.FlinkCDC /opt/datas/Flink-cdc2.0-1.0-SNAPSHOT-jar-with-dependencies.jar
  2.  
  3. Job has been submitted with JobID 137b680a6bb934e43568f14f6583b62c
手动执行savepoint

给当前程序创建保存点-savepoint

  1. [root@basenode flink-1.13.2]# bin/flink savepoint     e8e918c2517a777e817c630cf1d6b932    hdfs://192.168.1.161:8020/cdc-test/savepoint
  2. Triggering savepoint for job e8e918c2517a777e817c630cf1d6b932.
  3. Waiting for response...
  4. Savepoint completed. Path: hdfs://192.168.1.161:8020/cdc-test/savepoint/savepoint-e8e918-9ef094f349be
  5. You can resume your program from this savepoint with the run command.
  6. [root@basenode flink-1.13.2]#
界面停止 flink 程序

然后再mysql中添加数据

启动flink 程序
  1. [root@basenode flink-1.13.2]# bin/flink run -s hdfs://192.168.1.161:8020/cdc-test/savepoint/savepoint-e8e918-9ef094f349be -c  com.wud.cdc2.FlinkCDC /opt/datas/Flink-cdc2.0-1.0-SNAPSHOT-jar-with-dependencies.jar
  2. Job has been submitted with JobID 474a0da99820aa6025203f9806b9fcad

查看日志：

d99d2394a0360a1a7e54d47dc064502b.png
接下来 flink cdc 2.0 的自定义序列号函数

从上面可以看出flink cdc 的原始结构

  1. SourceRecord{sourcePartition={server=mysql_binlog_source}, 
  2.  sourceOffset={file=mysql-bin.000063, pos=154}}
  3.  ConnectRecord{topic='mysql_binlog_source.wudl-gmall.user_info', kafkaPartition=null, key=Struct{id=4000}, keySchema=Schema{mysql_binlog_source.wudl_gmall.user_info.Key:STRUCT}, value=Struct{after=Struct{id=4000,login_name=i0v0k9,nick_name=xxx,name=xxx,phone_num=137xxxxx,email=xxxx@qq.com,user_level=1,birthday=1969-12-04,gender=F,create_time=2020-12-04 23:28:45},source=Struct{version=1.4.1.Final,connector=mysql,name=mysql_binlog_source,ts_ms=0,snapshot=last,db=wudl-gmall,table=user_info,server_id=0,file=mysql-bin.000063,pos=154,row=0},op=c,ts_ms=1636255826014}, valueSchema=Schema{mysql_binlog_source.wudl_gmall.user_info.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}

我们可以自定义序列化：

  1. package com.wud.cdc2;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  5. import io.debezium.data.Envelope;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.util.Collector;
  9. import org.apache.kafka.connect.data.Field;
  10. import org.apache.kafka.connect.data.Schema;
  11. import org.apache.kafka.connect.data.Struct;
  12. import org.apache.kafka.connect.source.SourceRecord;
  13. import java.util.List;
  14.  
  15. public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
  16.     /**
  17.      *
  18.      * SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1636269821, file=mysql-bin.000063, pos=6442}} ConnectRecord{topic='mysql_binlog_source.test.Flink_iceberg', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{id=102,name=flinksql,age=25,dt=2021-11-08},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1636269821531,snapshot=last,db=test,table=Flink_iceberg,server_id=0,file=mysql-bin.000063,pos=6442,row=0},op=r,ts_ms=1636269821531}, valueSchema=Schema{mysql_binlog_source.test.Flink_iceberg.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  19.      *
  20.      *
  21.      *
  22.      *
  23.      *
  24.      * @param sourceRecord   返回一行数据
  25.      * @param collector 数据输出
  26.      * @throws Exception
  27.      */
  28.     @Override
  29.     public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  30.  
  31.         JSONObject  result = new JSONObject();
  32.         String topic = sourceRecord.topic();
  33.         String[] fields = topic.split("\\.");
  34.         result.put("db",fields[1]) ;
  35.         result.put("tableName",fields[2]);
  36.         // 获取before 数据
  37.         Struct value = (Struct) sourceRecord.value();
  38.         Struct before = value.getStruct("before");
  39.         JSONObject beforeJson = new JSONObject();
  40.         if (before !=null)
  41.         {
  42.             //获取列信息
  43.             Schema schema = before.schema();
  44.             List<Field> fieldList = schema.fields();
  45.             for (Field field:fieldList)
  46.             {
  47.                 beforeJson.put(field.name(),before.get(field));
  48.             }
  49.         }
  50.         result.put("before",beforeJson);
  51.         // 获取after 数据
  52.         Struct after = value.getStruct("after");
  53.         JSONObject afterJson = new JSONObject();
  54.         if (after !=null)
  55.         {
  56.             Schema schema = after.schema();
  57.             List<Field> afterFields = schema.fields();
  58.             for (Field field:afterFields)
  59.             {
  60.                 afterJson.put(field.name(),after.get(field));
  61.             }
  62.         }
  63.         result.put("after", afterJson);
  64.         //获取操作类型
  65.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  66.         result.put("op", operation);
  67.         //输出数据
  68.         collector.collect(result.toJSONString());
  69.     }
  70.     @Override
  71.     public TypeInformation<String> getProducedType() {
  72.         return  BasicTypeInfo.STRING_TYPE_INFO;
  73.     }
  74. }

调用flink cdc 的自定义函数

  1. package com.wud.cdc2;
  2.  
  3. import com.ververica.cdc.connectors.mysql.MySqlSource;
  4. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  6. import com.ververica.cdc.debezium.DebeziumSourceFunction;
  7. import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  8. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  9. import org.apache.flink.streaming.api.CheckpointingMode;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.environment.StreamPipelineOptions;
  14. import org.apache.flink.table.api.Table;
  15. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  16. import org.apache.flink.types.Row;
  17.  
  18. public class FlinkCDC {
  19.     public static void main(String[] args) throws Exception {
  20.  
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23. //        env.enableCheckpointing(5000L);
  24. //        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  25. //        // 设置任务关闭时候保留最后一次checkpoint 的数据
  26. //        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  27. //        // 指定ck 的自动重启策略
  28. //        env.setStateBackend(new FsStateBackend("hdfs://192.168.1.161:8020/cdc2.0-test/ck"));
  29. //        // 设置hdfs 的访问用户名
  30. //        System.setProperty("HADOOP_USER_NAME","hdfs");
  31.  
  32.         DebeziumSourceFunction<String> mySqlSource = MySqlSource.<String>builder()
  33.                 .hostname("192.168.1.180")
  34.                 .port(3306)
  35.                 .username("root")
  36.                 .password("123456")
  37.                 .databaseList("test")
  38.                 .tableList("test.Flink_iceberg")
  39. //                .deserializer(new StringDebeziumDeserializationSchema())
  40.                 .deserializer(new CustomerDeserialization())
  41.                 .startupOptions(StartupOptions.initial())
  42.                 .build();
  43.         DataStreamSource<String> dataStreamSource = env.addSource(mySqlSource);
  44.         dataStreamSource.print();
  45.         env.execute();
  46.     }
  47. }

新增一条数据可以看出 控制台输出结果：

{"op":"UPDATE","before":{"dt":"2021-11-07","name":"spark","id":104,"age":22},"after":{"dt":"2021-11-07","name":"spark02","id":104,"age":22},"db":"test","tableName":"Flink_iceberg"}

Flink CDC 2.0 SQL 可以做一个 ETL

需要注意的是必须要有主键 否则更新数据是新增一列，加主键后，更新数据不会新增。

数据库表结构:

  1. CREATE TABLE `Flink_iceberg` (
  2.   `id` bigint(64) DEFAULT NULL,
  3.   `name` varchar(64) DEFAULT NULL,
  4.   `age` int(20) DEFAULT NULL,
  5.   `dt` varchar(64) DEFAULT NULL
  6. ) ENGINE=InnoDB DEFAULT CHARSET=latin1

实现代码:

  1. package com.wud.cdc2;
  2.  
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4.  
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  6.  
  7. public class FlinkCdc20MysqlToMysql {
  8.     public static void main(String[] args) throws Exception {
  9.  
  10.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  11.         env.setParallelism(1);
  12.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  13.         String sourceSql = "CREATE TABLE IF NOT EXISTS mySqlSource (" +
  14.                 "id BIGINT primary key, " +
  15.                 "name string ," +
  16.                 "age int," +
  17.                 "dt string" +
  18.                 ") with ( " +
  19.                 " 'connector' = 'mysql-cdc', " +
  20.                 " 'scan.startup.mode' = 'latest-offset', " +
  21.                 " 'hostname' = '192.168.1.180', " +
  22.                 " 'port' = '3306', " +
  23.                 " 'username' = 'root', " +
  24.                 " 'password' = '123456', " +
  25.                 " 'database-name' = 'test', " +
  26.                 " 'table-name' = 'Flink_iceberg' " +
  27.                 ")";
  28.  
  29.         String sinkSql = " CREATE TABLE IF NOT EXISTS mySqlSink (" +
  30.                 "id BIGINT primary key , " +
  31.                 "name string ," +
  32.                 "age int," +
  33.                 "dt string" +
  34.                 ") with (" +
  35.                 " 'connector' = 'jdbc'," +
  36.                 " 'url' = 'jdbc:mysql://192.168.1.180:3306/test'," +
  37.                 "'table-name' = 'Flink_iceberg-cdc'," +
  38.                 " 'username' = 'root'," +
  39.                 " 'password' = '123456' " +
  40.                 " )";
  41.         tableEnv.executeSql(sourceSql);
  42.         tableEnv.executeSql(sinkSql);
  43.         tableEnv.executeSql("insert  into  mySqlSink select * from mySqlSource ");
  44. //        env.execute("FlinkCdc20MysqlToMysql");
  45.     }
  46. }
新增一条数据和更新数据显示
  1. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysqA","id":10012,"age":19},"db":"test","tableName":"Flink_iceberg"}
  2. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysqA","id":10012,"age":19},"db":"test","tableName":"Flink_iceberg"}
  3. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysql","id":10011,"age":19},"db":"test","tableName":"Flink_iceberg"}
  4. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysql","id":10011,"age":19},"db":"test","tableName":"Flink_iceberg"}
  5. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysqA","id":10012,"age":19},"db":"test","tableName":"Flink_iceberg"}
  6. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysqA3","id":10013,"age":19},"db":"test","tableName":"Flink_iceberg"}
  7. {"op":"READ","before":{},"after":{"dt":"2021-09-28","name":"flink-mysqA4","id":10014,"age":19},"db":"test","tableName":"Flink_iceberg"}
  8. {"op":"READ","before":{},"after":{"dt":"2021-09-24","name":"flink-mysql","id":10011,"age":19},"db":"test","tableName":"Flink_iceberg"}
  9. {"op":"READ","before":{},"after":{"dt":"2021-11-07","name":"flink","id":101,"age":20},"db":"test","tableName":"Flink_iceberg"}
  10. {"op":"READ","before":{},"after":{"dt":"2021-11-08","name":"flinksql","id":102,"age":25},"db":"test","tableName":"Flink_iceberg"}
  11. {"op":"READ","before":{},"after":{"dt":"2021-11-09","name":"flink-table","id":103,"age":21},"db":"test","tableName":"Flink_iceberg"}
  12. {"op":"READ","before":{},"after":{"dt":"2021-11-07","name":"spark","id":104,"age":22},"db":"test","tableName":"Flink_iceberg"}
  13. {"op":"READ","before":{},"after":{"dt":"2021-11-07","name":"hbase","id":105,"age":25},"db":"test","tableName":"Flink_iceberg"}
  14. {"op":"UPDATE","before":{"dt":"2021-11-07","name":"spark","id":104,"age":22},"after":{"dt":"2021-11-07","name":"spark02","id":104,"age":22},"db":"test","tableName":"Flink_iceberg"}
  15. {"op":"CREATE","before":{},"after":{"dt":"2021-11-07","name":"flinkcdc","id":106,"age":22},"db":"test","tableName":"Flink_iceberg"}

如果这个文章对你有帮助，不要忘记 「在看」 「点赞」 「收藏」 三连啊喂！

e5266a3421576ec11739ca9355b9c737.png

4515d189533933ea657127ab0eaea03f.png

2022年全网首发|大数据专家级技能模型与学习指南(胜天半子篇)

互联网最坏的时代可能真的来了

我在B站读大学，大数据专业

我们在学习Flink的时候，到底在学习什么？

193篇文章暴揍Flink，这个合集你需要关注一下

Flink生产环境TOP难题与优化，阿里巴巴藏经阁YYDS

Flink CDC我吃定了耶稣也留不住他！| Flink CDC线上问题小盘点

我们在学习Spark的时候，到底在学习什么？

在所有Spark模块中，我愿称SparkSQL为最强！

硬刚Hive | 4万字基础调优面试小总结

数据治理方法论和实践小百科全书

标签体系下的用户画像建设小指南

4万字长文 | ClickHouse基础&实践&调优全视角解析

【面试&个人成长】2021年过半，社招和校招的经验之谈

大数据方向另一个十年开启 |《硬刚系列》第一版完结

我写过的关于成长/面试/职场进阶的文章

当我们在学习Hive的时候在学习什么？「硬刚Hive续集」

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

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

  

闽ICP备14008679号