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

FlinkCDC初体验

作者：代码探险家 | 2024-08-21 07:55:49

赞

踩

FlinkCDC初体验

1、pom依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2.  
  3. <project xmlns="http://maven.apache.org/POM/4.0.0" 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.   <modelVersion>4.0.0</modelVersion>
  6.  
  7.   <groupId>org.example</groupId>
  8.   <artifactId>Flink1.17.2</artifactId>
  9.   <version>1.0-SNAPSHOT</version>
  10.  
  11.   <name>Flink1.17.2</name>
  12.   <!-- FIXME change it to the project's website -->
  13.   <url>http://www.example.com</url>
  14.  
  15.     <repositories>
  16.         <repository>
  17.             <id>oss.sonatype.org-snapshot</id>
  18.             <name>OSS Sonatype Snapshot Repository</name>
  19.             <url>http://oss.sonatype.org/content/repositories/snapshots</url>
  20.             <releases>
  21.                 <enabled>false</enabled>
  22.             </releases>
  23.             <snapshots>
  24.                 <enabled>true</enabled>
  25.             </snapshots>
  26.         </repository>
  27.         <repository>
  28.             <id>apache.snapshots</id>
  29.             <name>Apache Development Snapshot Repository</name>
  30.             <url>https://repository.apache.org/content/repositories/snapshots/</url>
  31.             <releases>
  32.                 <enabled>false</enabled>
  33.             </releases>
  34.             <snapshots>
  35.                 <enabled>true</enabled>
  36.             </snapshots>
  37.         </repository>
  38.     </repositories>
  39.   <properties>
  40.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  41.     <scala.version>2.12.7</scala.version>
  42.     <scala.binary.version>2.12</scala.binary.version>
  43.     <flink.version>1.17.2</flink.version>
  44.     <java.version>1.8</java.version>
  45.     <maven.compiler.source>1.8</maven.compiler.source>
  46.     <maven.compiler.target>1.8</maven.compiler.target>
  47.   </properties>
  48.  
  49.   <dependencies>
  50. <!--    <dependency>-->
  51. <!--      <groupId>com.alibaba.flink</groupId>-->
  52. <!--      <artifactId>datahub-connector</artifactId>-->
  53. <!--      <version>0.1-SNAPSHOT</version>-->
  54. <!--      <classifier>jar-with-dependencies</classifier>-->
  55. <!--    </dependency>-->
  56. <!--     https://mvnrepository.com/artifact/com.alibaba.ververica/ververica-connector-datahub -->
  57.         <dependency>
  58.           <groupId>com.alibaba.ververica</groupId>
  59.           <artifactId>ververica-connector-datahub</artifactId>
  60.           <version>1.17-vvr-8.0.8</version>
  61.             <scope>provided</scope>
  62.         </dependency>
  63.       <!-- https://mvnrepository.com/artifact/com.alibaba.ververica/ververica-connector-common -->
  64.       <dependency>
  65.           <groupId>com.alibaba.ververica</groupId>
  66.           <artifactId>ververica-connector-common</artifactId>
  67.           <version>1.17-vvr-8.0.8</version>
  68.           <scope>provided</scope>
  69.       </dependency>
  70.       <dependency>
  71.           <groupId>com.alibaba.ververica</groupId>
  72.           <artifactId>ververica-connector-continuous-odps</artifactId>
  73.           <version>1.17-vvr-8.0.8</version>
  74.           <scope>provided</scope>
  75.       </dependency>
  76.       <dependency>
  77.           <groupId>org.apache.flink</groupId>
  78.           <artifactId>flink-connector-base</artifactId>
  79.           <version>${flink.version}</version>
  80. <!--          <scope>provided</scope>-->
  81.       </dependency>
  82.  
  83.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->
  84.     <dependency>
  85.       <groupId>org.apache.flink</groupId>
  86.       <artifactId>flink-core</artifactId>
  87.       <version>${flink.version}</version>
  88. <!--        <scope>provided</scope>-->
  89.     </dependency>
  90.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
  91.     <dependency>
  92.       <groupId>org.apache.flink</groupId>
  93.       <artifactId>flink-java</artifactId>
  94.       <version>${flink.version}</version>
  95. <!--        <scope>provided</scope>-->
  96.     </dependency>
  97.       <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
  98.       <dependency>
  99.           <groupId>org.apache.flink</groupId>
  100.           <artifactId>flink-clients</artifactId>
  101.           <version>${flink.version}</version>
  102. <!--          <scope>provided</scope>-->
  103.       </dependency>
  104.  
  105.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
  106.     <dependency>
  107.       <groupId>org.apache.flink</groupId>
  108.       <artifactId>flink-streaming-java</artifactId>
  109.       <version>${flink.version}</version>
  110. <!--            <scope>provided</scope>-->
  111.     </dependency>
  112.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-runtime -->
  113.     <dependency>
  114.       <groupId>org.apache.flink</groupId>
  115.       <artifactId>flink-runtime</artifactId>
  116.       <version>${flink.version}</version>
  117. <!--            <scope>provided</scope>-->
  118.     </dependency>
  119.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-common -->
  120.     <dependency>
  121.       <groupId>org.apache.flink</groupId>
  122.       <artifactId>flink-table-common</artifactId>
  123.       <version>${flink.version}</version>
  124. <!--      <scope>provided</scope>-->
  125.     </dependency>
  126.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge -->
  127.     <dependency>
  128.       <groupId>org.apache.flink</groupId>
  129.       <artifactId>flink-table-api-java-bridge</artifactId>
  130.       <version>${flink.version}</version>
  131. <!--      <scope>provided</scope>-->
  132.     </dependency>
  133.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner -->
  134.     <dependency>
  135.       <groupId>org.apache.flink</groupId>
  136.       <artifactId>flink-table-planner_2.12</artifactId>
  137.       <version>${flink.version}</version>
  138. <!--      <scope>test</scope>-->
  139.     </dependency>
  140.  
  141.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-hive -->
  142.     <dependency>
  143.       <groupId>org.apache.flink</groupId>
  144.       <artifactId>flink-connector-hive_2.12</artifactId>
  145.       <version>${flink.version}</version>
  146.       <scope>provided</scope>
  147.     </dependency>
  148.  
  149.     <dependency>
  150.       <groupId>org.scala-lang</groupId>
  151.       <artifactId>scala-library</artifactId>
  152.       <version>2.12.7</version>
  153. <!--            <scope>provided</scope>-->
  154.     </dependency>
  155.     <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
  156.     <dependency>
  157.       <groupId>org.projectlombok</groupId>
  158.       <artifactId>lombok</artifactId>
  159.       <version>1.18.34</version>
  160.       <!--      <scope>provided</scope>-->
  161.     </dependency>
  162.     <dependency>
  163.       <groupId>mysql</groupId>
  164.       <artifactId>mysql-connector-java</artifactId>
  165.       <version>8.0.23</version>
  166.         <scope>provided</scope>
  167.     </dependency>
  168.       <!-- https://mvnrepository.com/artifact/com.ververica/flink-connector-debezium -->
  169.       <dependency>
  170.           <groupId>com.ververica</groupId>
  171.           <artifactId>flink-connector-debezium</artifactId>
  172.           <version>2.2.0</version>
  173. <!--          <scope>provided</scope>-->
  174.       </dependency>
  175.  
  176.     <dependency>
  177.       <groupId>com.ververica</groupId>
  178.       <artifactId>flink-connector-mysql-cdc</artifactId>
  179.       <version>2.2.0</version>
  180. <!--      <scope>provided</scope>-->
  181.     </dependency>
  182.     <dependency>
  183.           <groupId>com.ververica</groupId>
  184.           <artifactId>flink-sql-connector-mysql-cdc</artifactId>
  185.           <version>2.2.0</version>
  186. <!--          <scope>provided</scope>-->
  187.       </dependency>
  188.  
  189.  
  190.   </dependencies>
  191.  
  192.     <build>
  193.         <plugins>
  194.             <plugin>
  195.                 <groupId>org.apache.maven.plugins</groupId>
  196.                 <artifactId>maven-compiler-plugin</artifactId>
  197.                 <version>3.1</version>
  198.                 <configuration>
  199.                     <source>${java.version}</source>
  200.                     <target>${java.version}</target>
  201.                 </configuration>
  202.             </plugin>
  203.             <plugin>
  204.                 <groupId>org.apache.maven.plugins</groupId>
  205.                 <artifactId>maven-shade-plugin</artifactId>
  206.                 <version>3.2.4</version>
  207.                 <executions>
  208.                     <execution>
  209.                         <phase>package</phase>
  210.                         <goals>
  211.                             <goal>shade</goal>
  212.                         </goals>
  213.                         <configuration>
  214.                             <finalName>${project.artifactId}-${project.version}-flink-fat-jar</finalName>
  215.                             <transformers>
  216.                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  217.                                     <mainClass>com.tbea.cdc.FlinkCDCStreamExample</mainClass> <!-- 指定你的主类，例如 org.example.MyFlinkJob -->
  218.                                 </transformer>
  219.                             </transformers>
  220.                         </configuration>
  221.                     </execution>
  222.                 </executions>
  223.             </plugin>
  224.         </plugins>
  225.     </build>
  226. </project>

 2、FlinkCDC-SQL

  1. package com.tbea.cdc;
  2.  
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  5. import org.apache.flink.table.api.*;
  6.  
  7. public class FlinkCDCSqlExample {
  8.     public static void main(String[] args) throws Exception {
  9.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  10.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  11.  
  12.         // 创建HiveCatalog
  13. //        String name            = "myhive";
  14. //        String defaultDatabase = "mydatabase";
  15. //        String hiveConfDir      = "/path/to/your/hive/conf"; // 替换为你的Hive配置文件夹路径
  16. //        HiveCatalog hiveCatalog = new HiveCatalog(name, defaultDatabase, hiveConfDir);
  17. //        tableEnv.registerCatalog("myhive", hiveCatalog);
  18. //
  19. //        // 设置HiveCatalog为当前catalog
  20. //        tableEnv.useCatalog("myhive");
  21.  
  22.  
  23.         // 注册MySQL表
  24.         String sourceDDL = "CREATE TABLE source_table (" +
  25.                 " id INT," +
  26.                 " name STRING," +
  27.                 " age INT," +
  28.                 " addr STRING ,"+
  29.                 " create_time timestamp,"+
  30.                 " PRIMARY KEY (id) NOT ENFORCED" +
  31.                 ") WITH (" +
  32.                 " 'connector' = 'mysql-cdc'," +
  33.                 " 'hostname' = '192.168.140.1'," + // 替换为你的MySQL主机名
  34.                 " 'port' = '3306'," +
  35.                 " 'username' = 'flink'," + // 替换为你的MySQL用户名
  36.                 " 'password' = 'flink'," + // 替换为你的MySQL密码
  37.                 " 'database-name' = 'flinkcdc'," + // 替换为你的数据库名
  38.                 " 'table-name' = 'user_info'," + // 替换为你的表名
  39.                 //                " 'scan.startup.timestamp-millis' = '1000',"+
  40.                 " 'scan.incremental.snapshot.enabled' = 'true',"+
  41.                 " 'scan.incremental.snapshot.chunk.size' = '8096'," +
  42.                 " 'scan.startup.mode' = 'latest-offset'," +
  43.                 " 'debezium.snapshot.mode'= 'latest',"+
  44.                 " 'scan.newly-added-table.enabled' = 'true'"+
  45. //                " 'debezium.skipped.operations'='d'" +
  46.                 ")";
  47.  
  48.         tableEnv.executeSql(sourceDDL);
  49.  
  50.         // 查询并打印数据
  51.         TableResult result = tableEnv.executeSql("SELECT id,name,age,addr,create_time FROM source_table");
  52.  
  53.         result.print();
  54.         env.execute();
  55.     }
  56. }

运行结果：

3、FlinkCDC-DataStream

  1. package com.tbea.cdc;
  2.  
  3. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  4. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  6. import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  7. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  8. import org.apache.flink.api.common.restartstrategy.RestartStrategies;
  9. import org.apache.flink.api.common.time.Time;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12.  
  13. public class FlinkCDCStreamExample {
  14.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
  17.                 .hostname("192.168.140.1")
  18.                 .port(3306)
  19.                 // 设置捕获的数据库
  20.                 .databaseList("flinkcdc")
  21.                 // 设置捕获的表 [product, user, address]
  22.                 .tableList("flinkcdc.*")
  23.                 .username("flink")
  24.                 .password("flink")
  25.                 .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
  26.                 // 启用扫描新添加的表功能
  27.                 .scanNewlyAddedTableEnabled(true)
  28.                 .startupOptions(StartupOptions.latest())
  29.                 .build();
  30.         // 你的业务代码
  31.         // 设置 3s 的 checkpoint 间隔
  32.         env.setParallelism(1);
  33.         env.setRestartStrategy(RestartStrategies.failureRateRestart(1, Time.seconds(3L),Time.seconds(5L)));
  34.         env.enableCheckpointing(3000);
  35.  
  36.         DataStreamSource<String> dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "Mysql-Source");
  37.         dataStreamSource.print();
  38.  
  39.  
  40.         env.execute("Print MySQL Snapshot + Binlog");
  41.     }
  42. }

 运行结果：

  1. {"before":null,"after":{"id":22,"name":"杨延昭","age":45,"addr":"辽宁","create_time":1723798965000},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1723770167000,"snapshot":"false","db":"flinkcdc","sequence":null,"table":"user_info","server_id":1,"gtid":null,"file":"YANGYINGCHUN-bin.000004","pos":8640,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1723770167469,"transaction":null}
  2.  
  3. {"before":{"id":17,"name":"储熊","age":28,"addr":"张北","create_time":null},"after":{"id":17,"name":"储熊","age":28,"addr":"张北","create_time":1723798410000},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1723769613000,"snapshot":"false","db":"flinkcdc","sequence":null,"table":"user_info","server_id":1,"gtid":null,"file":"YANGYINGCHUN-bin.000004","pos":6540,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1723769613737,"transaction":null}
  4.  
  5. {"before":{"id":8,"name":"赵刚","age":33,"addr":"延安","create_time":1724230872000},"after":null,"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1723770354000,"snapshot":"false","db":"flinkcdc","sequence":null,"table":"user_info","server_id":1,"gtid":null,"file":"YANGYINGCHUN-bin.000004","pos":8968,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1723770354642,"transaction":null}
声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

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

  

闽ICP备14008679号