0
0

Flink CDC 读取MySQL的数据

1、前提背景准备

Flink在1.11之后就已经支持从MySQL增量读取Binlog日志的方式。

pom文件如下：

  1. <properties>
  2.     <scala.binary.version>2.11</scala.binary.version>
  3.     <scala.version>2.11.12</scala.version>
  4.     <flink.version>1.12.0</flink.version>
  5.     <fastjson.verson>1.2.72</fastjson.verson>
  6.     <lombok.version>1.18.6</lombok.version>
  7.     <kafka.version>2.3.0</kafka.version>
  9. </properties>
  11. <dependencies>
  13.     <dependency>
  14.         <groupId>com.github.shyiko</groupId>
  15.         <artifactId>mysql-binlog-connector-java</artifactId>
  16.         <version>0.21.0</version>
  17.     </dependency>
  19.     <dependency>
  20.         <groupId>org.apache.kafka</groupId>
  21.         <artifactId>kafka-clients</artifactId>
  22.         <version>${kafka.version}</version>
  23.     </dependency>
  25.     <dependency>
  26.         <groupId>org.apache.flink</groupId>
  27.         <artifactId>flink-connector-kafka_2.11</artifactId>
  28.         <version>${flink.version}</version>
  29.         <exclusions>
  30.             <exclusion>
  31.                 <groupId>log4j</groupId>
  32.                 <artifactId>*</artifactId>
  33.             </exclusion>
  34.             <exclusion>
  35.                 <groupId>org.slf4j</groupId>
  36.                 <artifactId>slf4j-log4j12</artifactId>
  37.             </exclusion>
  38.         </exclusions>
  39.     </dependency>
  41.     <dependency>
  42.         <groupId>org.projectlombok</groupId>
  43.         <artifactId>lombok</artifactId>
  44.         <version>${lombok.version}</version>
  45.         <scope>provided</scope>
  46.     </dependency>
  48.     <dependency>
  49.         <groupId>com.alibaba</groupId>
  50.         <artifactId>fastjson</artifactId>
  51.         <version>${fastjson.verson}</version>
  52.     </dependency>
  54.     <dependency>
  55.         <groupId>com.alibaba.ververica</groupId>
  56.         <artifactId>flink-connector-mysql-cdc</artifactId>
  57.         <version>1.4.0</version>
  58.     </dependency>
  60.     <dependency>
  61.         <groupId>org.apache.flink</groupId>
  62.         <artifactId>flink-java</artifactId>
  63.         <version>${flink.version}</version>
  64.         <!--<scope>provided</scope>-->
  65.         <exclusions>
  66.             <exclusion>
  67.                 <groupId>log4j</groupId>
  68.                 <artifactId>*</artifactId>
  69.             </exclusion>
  70.             <exclusion>
  71.                 <groupId>org.slf4j</groupId>
  72.                 <artifactId>slf4j-log4j12</artifactId>
  73.             </exclusion>
  74.         </exclusions>
  75.     </dependency>
  77.     <dependency>
  78.         <groupId>org.apache.flink</groupId>
  79.         <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  80.         <version>${flink.version}</version>
  81.         <!--<scope>provided</scope>-->
  82.         <exclusions>
  83.             <exclusion>
  84.                 <groupId>log4j</groupId>
  85.                 <artifactId>*</artifactId>
  86.             </exclusion>
  87.             <exclusion>
  88.                 <groupId>org.slf4j</groupId>
  89.                 <artifactId>slf4j-log4j12</artifactId>
  90.             </exclusion>
  91.             <exclusion>
  92.                 <groupId>com.google.code.findbugs</groupId>
  93.                 <artifactId>jsr305</artifactId>
  94.             </exclusion>
  95.             <exclusion>
  96.                 <groupId>org.apache.flink</groupId>
  97.                 <artifactId>force-shading</artifactId>
  98.             </exclusion>
  99.         </exclusions>
  100.     </dependency>
  102.     <dependency>
  103.         <groupId>org.apache.flink</groupId>
  104.         <artifactId>flink-clients_${scala.binary.version}</artifactId>
  105.         <version>${flink.version}</version>
  106.         <!--<scope>provided</scope>-->
  107.     </dependency>
  109.     <dependency>
  110.         <groupId>org.apache.flink</groupId>
  111.         <artifactId>flink-scala_${scala.binary.version}</artifactId>
  112.         <version>${flink.version}</version>
  113.     </dependency>
  115.     <dependency>
  116.         <groupId>org.apache.flink</groupId>
  117.         <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
  118.         <version>${flink.version}</version>
  119.         <exclusions>
  120.             <exclusion>
  121.                 <groupId>log4j</groupId>
  122.                 <artifactId>*</artifactId>
  123.             </exclusion>
  124.             <exclusion>
  125.                 <groupId>org.slf4j</groupId>
  126.                 <artifactId>slf4j-log4j12</artifactId>
  127.             </exclusion>
  128.         </exclusions>
  129.     </dependency>
  131. </dependencies>

2、全量读取某个数据库中的所有库中的所有表的Binlog方式代码如下：

  1. public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         Properties properties = new Properties();
  7.         // 配置 Debezium在初始化快照的时候（扫描历史数据的时候） =》 不要锁表
  8.         properties.setProperty("debezium.snapshot.locking.mode", "none");
  10.         env.enableCheckpointing(TimeUnit.SECONDS.toMillis(5));
  12.         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
  14.         //最大同时存在的ck数 和设置的间隔时间有一个就行
  15.         checkpointConfig.setMaxConcurrentCheckpoints(1);
  16.         //超时时间
  17.         checkpointConfig.setCheckpointTimeout(TimeUnit.SECONDS.toMillis(5));
  18.         //2.3 指定从CK自动重启策略
  19.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 5000L));
  20.         //2.4 设置任务关闭的时候保留最后一次CK数据
  21.         checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  23.         DebeziumSourceFunction<String> MysqlSource = MySQLSource.<String>builder()
  24.                 .hostname("100.21.112.11")
  25.                 .port(3306)
  26.                 .deserializer(new MyDeserializationSchema()) //去参数里面找找实现类
  27.                 .username("root")
  28.                 .password("xxxx")
  29.                 .startupOptions(StartupOptions.latest())// 读取binlog策略 这个启动选项有五种
  30.                 .debeziumProperties(properties) //配置不要锁表 但是数据一致性不是精准一次 会变成最少一次
  31.                 .build();
  33.         /*
  34.          *  .startupOptions(StartupOptions.latest()) 参数配置
  35.          *  1.initial() 全量扫描并且继续读取最新的binlog 最佳实践是第一次使用这个
  36.          *  2.earliest() 从binlog的开头开始读取 就是啥时候开的binlog就从啥时候读
  37.          *  3.latest() 从最新的binlog开始读取
  38.          *  4.specificOffset(String specificOffsetFile, int specificOffsetPos) 指定offset读取
  39.          *  5.timestamp(long startupTimestampMillis) 指定时间戳读取
  40.          * */
  42.         env.addSource(MysqlSource).print();
  44.         env.execute("flink-cdc");
  46.     }
  48.     public static class MyDeserializationSchema implements DebeziumDeserializationSchema<String> {
  50.         @Override
  51.         public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  52.             Struct value = (Struct) sourceRecord.value();
  53.             Struct after = value.getStruct("after");
  54.             Struct source = value.getStruct("source");
  56.             String db = source.getString("db");//库名
  57.             String table = source.getString("table");//表名
  59.             //获取操作类型 直接将参数穿进去 会自己解析出来 里面是个enum对应每个操作
  60.             /* READ("r"),
  61.                CREATE("c"),
  62.                 UPDATE("u"),
  63.                 DELETE("d");*/
  64.             Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  65.             String opstr = operation.toString().toLowerCase();
  66.             //类型修正 会把insert识别成create
  67.             if (opstr.equals("create")) {
  68.                 opstr = "insert";
  69.             }
  71.             //获取after结构体里面的表数据，封装成json输出
  72.             JSONObject json1 = new JSONObject();
  73.             JSONObject json2 = new JSONObject();
  74.             //加个判空
  75.             if (after != null) {
  76.                 List<Field> data = after.schema().fields(); //获取结构体
  77.                 for (Field field : data) {
  78.                     String name = field.name(); //结构体的名字
  79.                     Object value2 = after.get(field);//结构体的字段值
  80.                     //放进json2里面去 json2放到json1里面去
  81.                     json2.put(name, value2);
  82.                 }
  83.             }
  85.             //整理成大json串输出
  86.             json1.put("db", db);
  87.             json1.put("table", table);
  88.             json1.put("data", json2);
  89.             json1.put("type", opstr);
  91.             collector.collect(json1.toJSONString());
  93.         }
  95.         @Override
  96.         public TypeInformation<String> getProducedType() {
  97.             return TypeInformation.of(String.class);
  98.         }
  99.     }

3、全量读取某个数据库指定DB中的所有表

可以在build之前 ，添加一个

  1. databaseList，用来指定特定的DB
  1. public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         Properties properties = new Properties();
  7.         // 配置 Debezium在初始化快照的时候（扫描历史数据的时候） =》 不要锁表
  8.         properties.setProperty("debezium.snapshot.locking.mode", "none");
  10.         env.enableCheckpointing(TimeUnit.SECONDS.toMillis(5));
  12.         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
  14.         //最大同时存在的ck数 和设置的间隔时间有一个就行
  15.         checkpointConfig.setMaxConcurrentCheckpoints(1);
  16.         //超时时间
  17.         checkpointConfig.setCheckpointTimeout(TimeUnit.SECONDS.toMillis(5));
  18.         //2.3 指定从CK自动重启策略
  19.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 5000L));
  20.         //2.4 设置任务关闭的时候保留最后一次CK数据
  21.         checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  23.         DebeziumSourceFunction<String> MysqlSource = MySQLSource.<String>builder()
  24.                 .hostname("100.21.112.11")
  25.                 .port(3306)
  26.                 .deserializer(new MyDeserializationSchema()) //去参数里面找找实现类
  27.                 .username("root")
  28.                 .password("xxxx")
  29.                 .databaseList("horse") // 指定某个特定的库
  30.                 .startupOptions(StartupOptions.latest())// 读取binlog策略 这个启动选项有五种
  31.                 .debeziumProperties(properties) //配置不要锁表 但是数据一致性不是精准一次 会变成最少一次
  32.                 .build();
  34.         /*
  35.          *  .startupOptions(StartupOptions.latest()) 参数配置
  36.          *  1.initial() 全量扫描并且继续读取最新的binlog 最佳实践是第一次使用这个
  37.          *  2.earliest() 从binlog的开头开始读取 就是啥时候开的binlog就从啥时候读
  38.          *  3.latest() 从最新的binlog开始读取
  39.          *  4.specificOffset(String specificOffsetFile, int specificOffsetPos) 指定offset读取
  40.          *  5.timestamp(long startupTimestampMillis) 指定时间戳读取
  41.          * */
  43.         env.addSource(MysqlSource).print();
  45.         env.execute("flink-cdc");
  47.     }
  49.     public static class MyDeserializationSchema implements DebeziumDeserializationSchema<String> {
  51.         @Override
  52.         public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  53.             Struct value = (Struct) sourceRecord.value();
  54.             Struct after = value.getStruct("after");
  55.             Struct source = value.getStruct("source");
  57.             String db = source.getString("db");//库名
  58.             String table = source.getString("table");//表名
  60.             //获取操作类型 直接将参数穿进去 会自己解析出来 里面是个enum对应每个操作
  61.             /* READ("r"),
  62.                CREATE("c"),
  63.                 UPDATE("u"),
  64.                 DELETE("d");*/
  65.             Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  66.             String opstr = operation.toString().toLowerCase();
  67.             //类型修正 会把insert识别成create
  68.             if (opstr.equals("create")) {
  69.                 opstr = "insert";
  70.             }
  72.             //获取after结构体里面的表数据，封装成json输出
  73.             JSONObject json1 = new JSONObject();
  74.             JSONObject json2 = new JSONObject();
  75.             //加个判空
  76.             if (after != null) {
  77.                 List<Field> data = after.schema().fields(); //获取结构体
  78.                 for (Field field : data) {
  79.                     String name = field.name(); //结构体的名字
  80.                     Object value2 = after.get(field);//结构体的字段值
  81.                     //放进json2里面去 json2放到json1里面去
  82.                     json2.put(name, value2);
  83.                 }
  84.             }
  86.             //整理成大json串输出
  87.             json1.put("db", db);
  88.             json1.put("table", table);
  89.             json1.put("data", json2);
  90.             json1.put("type", opstr);
  92.             collector.collect(json1.toJSONString());
  94.         }
  96.         @Override
  97.         public TypeInformation<String> getProducedType() {
  98.             return TypeInformation.of(String.class);
  99.         }
  100.     }

4、全量读取某个数据库指定DB中的指定表

可以在build之前 ，添加一个

  1. tableList，用来指定特定的DB中的特定表
  1. public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         Properties properties = new Properties();
  7.         // 配置 Debezium在初始化快照的时候（扫描历史数据的时候） =》 不要锁表
  8.         properties.setProperty("debezium.snapshot.locking.mode", "none");
  10.         env.enableCheckpointing(TimeUnit.SECONDS.toMillis(5));
  12.         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
  14.         //最大同时存在的ck数 和设置的间隔时间有一个就行
  15.         checkpointConfig.setMaxConcurrentCheckpoints(1);
  16.         //超时时间
  17.         checkpointConfig.setCheckpointTimeout(TimeUnit.SECONDS.toMillis(5));
  18.         //2.3 指定从CK自动重启策略
  19.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 5000L));
  20.         //2.4 设置任务关闭的时候保留最后一次CK数据
  21.         checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  23.         DebeziumSourceFunction<String> MysqlSource = MySQLSource.<String>builder()
  24.                 .hostname("100.21.112.11")
  25.                 .port(3306)
  26.                 .deserializer(new MyDeserializationSchema()) //去参数里面找找实现类
  27.                 .username("root")
  28.                 .password("xxxx")
  29.                 .databaseList("horse") // 指定某个特定的库
  30.                 .tableList("horse.t_dri_info") //指定特定的表
  31.                 .startupOptions(StartupOptions.latest())// 读取binlog策略 这个启动选项有五种
  32.                 .debeziumProperties(properties) //配置不要锁表 但是数据一致性不是精准一次 会变成最少一次
  33.                 .build();
  35.         /*
  36.          *  .startupOptions(StartupOptions.latest()) 参数配置
  37.          *  1.initial() 全量扫描并且继续读取最新的binlog 最佳实践是第一次使用这个
  38.          *  2.earliest() 从binlog的开头开始读取 就是啥时候开的binlog就从啥时候读
  39.          *  3.latest() 从最新的binlog开始读取
  40.          *  4.specificOffset(String specificOffsetFile, int specificOffsetPos) 指定offset读取
  41.          *  5.timestamp(long startupTimestampMillis) 指定时间戳读取
  42.          * */
  44.         env.addSource(MysqlSource).print();
  46.         env.execute("flink-cdc");
  48.     }
  50.     public static class MyDeserializationSchema implements DebeziumDeserializationSchema<String> {
  52.         @Override
  53.         public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  54.             Struct value = (Struct) sourceRecord.value();
  55.             Struct after = value.getStruct("after");
  56.             Struct source = value.getStruct("source");
  58.             String db = source.getString("db");//库名
  59.             String table = source.getString("table");//表名
  61.             //获取操作类型 直接将参数穿进去 会自己解析出来 里面是个enum对应每个操作
  62.             /* READ("r"),
  63.                CREATE("c"),
  64.                 UPDATE("u"),
  65.                 DELETE("d");*/
  66.             Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  67.             String opstr = operation.toString().toLowerCase();
  68.             //类型修正 会把insert识别成create
  69.             if (opstr.equals("create")) {
  70.                 opstr = "insert";
  71.             }
  73.             //获取after结构体里面的表数据，封装成json输出
  74.             JSONObject json1 = new JSONObject();
  75.             JSONObject json2 = new JSONObject();
  76.             //加个判空
  77.             if (after != null) {
  78.                 List<Field> data = after.schema().fields(); //获取结构体
  79.                 for (Field field : data) {
  80.                     String name = field.name(); //结构体的名字
  81.                     Object value2 = after.get(field);//结构体的字段值
  82.                     //放进json2里面去 json2放到json1里面去
  83.                     json2.put(name, value2);
  84.                 }
  85.             }
  87.             //整理成大json串输出
  88.             json1.put("db", db);
  89.             json1.put("table", table);
  90.             json1.put("data", json2);
  91.             json1.put("type", opstr);
  93.             collector.collect(json1.toJSONString());
  95.         }
  97.         @Override
  98.         public TypeInformation<String> getProducedType() {
  99.             return TypeInformation.of(String.class);
  100.         }
  101.     }
标签： mysql flink 数据库
本文转载自: https://blog.csdn.net/Aaron_ch/article/details/122133198
版权归原作者 一个肉团子 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Flink CDC 读取MySQL的数据”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？