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

FlinkCDC读取MySQL并写入Kafka案例(com.alibaba.ververica)

作者：Li_阴宅 | 2024-07-16 21:57:46

赞

踩

com.alibaba.ververica

场景应用：将MySQL的变化数据转为实时流输出到Kafka中。

注意版本问题，版本不同可能会出现异常，以下版本测试没问题：

flink1.12.7

flink-connector-mysql-cdc 1.3.0(com.alibaba.ververica) (测试时使用1.2.0版本时会出现空指针错误)

1. MySQL的配置

 在/etc/my.cnf文件中，【mysqld】下面添加以下配置：

binlog-do-db  是指定要监控的数据库，如果是多个数据库，每个数据库需要单独一行设置。

 修改完成后，需要重启数据库，并检查binlog有没有生成。

补充几个其他的配置：

  1. 1、修改配置
  2.     [mysqld]
  3.     # 前面还有其他配置
  4.     # 添加的部分
  5.     server-id = 12345
  6.     log-bin = mysql-bin
  7.     # 必须为ROW
  8.     binlog_format = ROW
  9.     # 必须为FULL，MySQL-5.7后才有该参数
  10.     binlog_row_image  = FULL
  11.     expire_logs_days  = 15
  12.  
  13. 2、验证
  14. SHOW VARIABLES LIKE '%binlog%';
  15.     
  16. 3、设置权限
  17.     -- 设置拥有同步权限的用户
  18.     CREATE USER 'flinkuser' IDENTIFIED BY 'flinkpassword';
  19.     -- 赋予同步相关权限
  20.     GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flinkuser';
  21.     
  22.     创建用户并赋予权限成功后，使用该用户登录MySQL，可以使用以下命令查看主从同步相关信息
  23.     SHOW MASTER STATUS
  24.     SHOW SLAVE STATUS
  25.     SHOW BINARY LOGS

2. FlinkCDC的开发

从这里开始建立flink工程项目，以下项目flink版本为1.12.7，scala版本用的2.12。

大概的思考步骤如下：

1) 获取执行环境

2)开启检查点ck (重点)

3)通过flinkcdc构建sourceFunction，并读取数据 (重点)

4)在执行环境中添加3)中构建的source

5)配置kafka生产者环境(重点)

6)在执行环境中增加5)中的Sink

7)启动任务

项目结构(gmall-realtime)如下：

2.1 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>gmall-flink-2021</artifactId>
  7.         <groupId>com.king</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>gmall-flink-cdc</artifactId>
  13.     <version>1.0</version>
  14.  
  15.     <properties>
  16.         <java.version>1.8</java.version>
  17.         <maven.compiler.source>${java.version}</maven.compiler.source>
  18.         <maven.compiler.target>${java.version}</maven.compiler.target>
  19.         <flink.version>1.12.7</flink.version>
  20.         <scala.version>2.12</scala.version>
  21.         <hadoop.version>3.1.3</hadoop.version>
  22.     </properties>
  23.  
  24.     <dependencies>
  25.         <dependency>
  26.             <groupId>org.apache.flink</groupId>
  27.             <artifactId>flink-java</artifactId>
  28.             <version>${flink.version}</version>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>org.apache.flink</groupId>
  32.             <artifactId>flink-streaming-java_${scala.version}</artifactId>
  33.             <version>${flink.version}</version>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>org.apache.flink</groupId>
  37.             <artifactId>flink-connector-kafka_${scala.version}</artifactId>
  38.             <version>${flink.version}</version>
  39.         </dependency>
  40.         <dependency>
  41.             <groupId>org.apache.flink</groupId>
  42.             <artifactId>flink-clients_${scala.version}</artifactId>
  43.             <version>${flink.version}</version>
  44.         </dependency>
  45.         <dependency>
  46.             <groupId>org.apache.flink</groupId>
  47.             <artifactId>flink-cep_${scala.version}</artifactId>
  48.             <version>${flink.version}</version>
  49.         </dependency>
  50.         <dependency>
  51.             <groupId>org.apache.flink</groupId>
  52.             <artifactId>flink-json</artifactId>
  53.             <version>${flink.version}</version>
  54.         </dependency>
  55.         <dependency>
  56.             <groupId>com.alibaba</groupId>
  57.             <artifactId>fastjson</artifactId>
  58.             <version>1.2.68</version>
  59.         </dependency>
  60.         <!--如果保存检查点到 hdfs 上，需要引入此依赖-->
  61.         <dependency>
  62.             <groupId>org.apache.hadoop</groupId>
  63.             <artifactId>hadoop-client</artifactId>
  64.             <version>${hadoop.version}</version>
  65.         </dependency>
  66.         <dependency>
  67.             <groupId>mysql</groupId>
  68.             <artifactId>mysql-connector-java</artifactId>
  69.             <version>8.0.16</version>
  70.         </dependency>
  71.  
  72.         <dependency>
  73.             <groupId>org.apache.kafka</groupId>
  74.             <artifactId>kafka-clients</artifactId>
  75.             <version>2.7.0</version>
  76.         </dependency>
  77.         <dependency>
  78.             <groupId>com.alibaba.ververica</groupId>
  79.             <artifactId>flink-connector-mysql-cdc</artifactId>
  80.             <version>1.3.0</version>
  81.         </dependency>
  82.         <!-- https://mvnrepository.com/artifact/com.ververica/flink-connector-mysql-cdc -->
  83. <!--        <dependency>  该包仅支持flink1.13版本及以上-->
  84. <!--            <groupId>com.ververica</groupId>-->
  85. <!--            <artifactId>flink-connector-mysql-cdc</artifactId>-->
  86. <!--            <version>2.1.1</version>-->
  87. <!--        </dependency>-->
  88.         <dependency>
  89.             <groupId>org.projectlombok</groupId>
  90.             <artifactId>lombok</artifactId>
  91.             <version>1.18.20</version>
  92.         </dependency>
  93.         <!--Flink 默认使用的是 slf4j 记录日志，相当于一个日志的接口,我们这里使用 log4j 作为
  94.         具体的日志实现-->
  95.         <dependency>
  96.             <groupId>org.slf4j</groupId>
  97.             <artifactId>slf4j-api</artifactId>
  98.             <version>1.7.32</version>
  99.         </dependency>
  100.         <dependency>
  101.             <groupId>org.slf4j</groupId>
  102.             <artifactId>slf4j-log4j12</artifactId>
  103.             <version>1.7.32</version>
  104.         </dependency>
  105.         <dependency>
  106.             <groupId>org.apache.logging.log4j</groupId>
  107.             <artifactId>log4j-to-slf4j</artifactId>
  108.             <version>2.17.1</version>
  109.         </dependency>
  110.     </dependencies>
  111.  
  112.  
  113.     <build>
  114. <!--        <sourceDirectory>${project.basedir}/src/main/scala</sourceDirectory>-->
  115. <!--        <resources>-->
  116. <!--            <resource>-->
  117. <!--                <directory>${project.basedir}/src/main/resources</directory>-->
  118. <!--            </resource>-->
  119. <!--        </resources>-->
  120.         <plugins>
  121.             <plugin>
  122.                 <groupId>org.apache.maven.plugins</groupId>
  123.                 <artifactId>maven-assembly-plugin</artifactId>
  124.                 <version>3.0.0</version>
  125.                 <configuration>
  126.                     <descriptorRefs>
  127.                         <descriptorRef>jar-with-dependencies</descriptorRef>
  128.                     </descriptorRefs>
  129.                 </configuration>
  130.                 <executions>
  131.                     <execution>
  132.                         <id>make-assembly</id>
  133.                         <phase>package</phase>
  134.                         <goals>
  135.                             <goal>single</goal>
  136.                         </goals>
  137.                     </execution>
  138.                 </executions>
  139.             </plugin>
  140.         </plugins>
  141.     </build>
  142.  
  143.  
  144. </project>
pom.xml

注意一点：如果使用java开发，可以直接编译成功。但是我这里全部使用scala开发，所以需要在pom文件配置额外的插件，否则打包scala项目会不成功。

  1. <plugins>
  2.             <plugin>
  3.                 <!-- ！！必须有这个插件，才可以编译scala代码找到主类，版本我是网上搞来的 -->
  4.                 <groupId>net.alchim31.maven</groupId>
  5.                 <artifactId>scala-maven-plugin</artifactId>
  6.                 <version>3.2.2</version>
  7.                 <executions>
  8.                     <execution>
  9.                         <id>compile-scala</id>
  10.                         <phase>compile</phase>
  11.                         <goals>
  12.                             <goal>add-source</goal>
  13.                             <goal>compile</goal>
  14.                         </goals>
  15.                     </execution>
  16.                     <execution>
  17.                         <id>test-compile-scala</id>
  18.                         <phase>test-compile</phase>
  19.                         <goals>
  20.                             <goal>add-source</goal>
  21.                             <goal>testCompile</goal>
  22.                         </goals>
  23.                     </execution>
  24.                 </executions>
  25.             </plugin>
  26.         </plugins>
plugins
2.2 读取MySQL

Flinkcdc.scala中：

通过引入的flink-connector-mysql-cdc已经提供了读取MySQL的工具类。

  1.     val sourceFunction = MySQLSource.builder[String]()
  2.         .hostname("hadoop200")
  3.         .port(3306)
  4.         .username("root")
  5.         .password("root")
  6.         .databaseList("gmall-210325-flink")


        //如果不添加该参数，则消费指定数据库中所有表的数据        //如果添加，则需要按照 数据库名.表名 的格式指定，多个表使用逗号隔开//        .tableList("gmall-210325-flink.base_trademark")        .deserializer(new CustomerDeseriallization())

new CustomerDeseriallization() 是自定义的读取的MySQL的数据输出格式，如果不指定，系统也有个new StringDebeziumDeserializationSchema()可以使用。
2.3 自定义从MySQL读取的数据的输出格式
CustomerDeseriallization类

  1. package com.king.app.function
  2.  
  3. import com.alibaba.fastjson.JSONObject
  4. import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema
  5. import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
  6. import org.apache.flink.util.Collector
  7. import org.apache.kafka.connect.data.{Schema, Struct}
  8. import org.apache.kafka.connect.source.SourceRecord
  9.  
  10.  
  11. /**
  12.  * @Author: KingWang
  13.  * @Date: 2021/12/29  
  14.  * @Desc:
  15.  **/
  16. class CustomerDeseriallization extends DebeziumDeserializationSchema[String]{
  17.  
  18.   /**
  19.    * 封装的数据：
  20.    * {
  21.    *   "database":"",
  22.    *   "tableName":"",
  23.    *   "type":"c r u d",
  24.    *   "before":"",
  25.    *   "after":"",
  26.    *   "ts": ""
  27.    *
  28.    * }
  29.    *
  30.    * @param sourceRecord
  31.    * @param collector
  32.    */
  33.   override def deserialize(sourceRecord: SourceRecord, collector: Collector[String]): Unit = {
  34.     //1. 创建json对象用于保存最终数据
  35.     val result = new JSONObject()
  36.  
  37.  
  38.     val value:Struct = sourceRecord.value().asInstanceOf[Struct]
  39.     //2. 获取库名&表名
  40.     val source:Struct = value.getStruct("source")
  41.     val database = source.getString("db")
  42.     val table = source.getString("table")
  43.  
  44.     //3. 获取before
  45.     val before = value.getStruct("before")
  46.     val beforeObj = if(before != null)  getJSONObjectBySchema(before.schema(),before) else new JSONObject()
  47.  
  48.  
  49.     //4. 获取after
  50.     val after = value.getStruct("after")
  51.     val afterObj = if(after != null) getJSONObjectBySchema(after.schema(),after) else new JSONObject()
  52.  
  53.     //5. 获取操作类型
  54.     val op:String = value.getString("op")
  55.  
  56.     //6. 获取操作时间
  57.     val ts = source.getInt64("ts_ms")
  58. //    val ts = value.getInt64("ts_ms")
  59.  
  60.  
  61.     //7. 拼接结果
  62.     result.put("database", database)
  63.     result.put("table", table)
  64.     result.put("type", op)
  65.     result.put("before", beforeObj)
  66.     result.put("after", afterObj)
  67.     result.put("ts", ts)
  68.  
  69.     collector.collect(result.toJSONString)
  70.  
  71.   }
  72.  
  73.  
  74.   override def getProducedType: TypeInformation[String] = {
  75.     BasicTypeInfo.STRING_TYPE_INFO
  76.   }
  77.  
  78.  
  79.   //从Schema中获取字段和值
  80.   def getJSONObjectBySchema(schema:Schema,struct:Struct):JSONObject = {
  81.     val fields = schema.fields()
  82.     var jsonBean = new JSONObject()
  83.     val iter = fields.iterator()
  84.     while(iter.hasNext){
  85.       val field = iter.next()
  86.       val key = field.name()
  87.       val value = struct.get(field)
  88.       jsonBean.put(key,value)
  89.     }
  90.     jsonBean
  91.   }
  92.  
  93. }
CustomerDeseriallization
2.4 写入到Kafka
  1. package com.king.util
  2.  
  3. import org.apache.flink.api.common.serialization.{SerializationSchema, SimpleStringSchema}
  4. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
  5.  
  6. /**
  7.  * @Author: KingWang
  8.  * @Date: 2022/1/1  
  9.  * @Desc:
  10.  **/
  11. object MyKafkaUtil {
  12.  
  13.   val broker_list = "hadoop200:9092,hadoop201:9092,hadoop202:9092"
  14.  
  15.   def getKafkaProducer(topic:String):FlinkKafkaProducer[String] =
  16.     new FlinkKafkaProducer[String](broker_list,topic,new SimpleStringSchema())
  17.  
  18.  
  19. }
MyKafkaUtil

FlinkCDC.scala的完整代码如下：

  1. package com.king.app.ods
  2.  
  3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource
  4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions
  5. import com.king.app.function.CustomerDeseriallization
  6. import com.king.util.MyKafkaUtil
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
  8.  
  9. /**
  10.  * @Author: KingWang
  11.  * @Date: 2021/12/26  
  12.  * @Desc:
  13.  **/
  14. object FlinkCDC {
  15.  
  16.   def main(args: Array[String]): Unit = {
  17.  
  18.     //1. 获取执行环境
  19.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  20.     env.setParallelism(1)
  21.  
  22.     //1.1 开启ck并指定状态后端fs
  23. //    env.setStateBackend(new FsStateBackend("hdfs://hadoop200:8020/gmall-flink-210325/ck"))
  24. //      .enableCheckpointing(10000L) //头尾间隔：每10秒触发一次ck
  25. //      env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)  //
  26. //      env.getCheckpointConfig.setCheckpointTimeout(10000L)
  27. //    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
  28. //    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(3000l)  //尾和头间隔时间3秒
  29.  
  30. //    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));
  31.  
  32.  
  33.  
  34.     //2. 通过flinkCDC构建SourceFunction并读取数据
  35.     val sourceFunction = MySQLSource.builder[String]()
  36.         .hostname("hadoop200")
  37.         .port(3306)
  38.         .username("root")
  39.         .password("root")
  40.         .databaseList("gmall-210325-flink")
  41.  
  42.         //如果不添加该参数，则消费指定数据库中所有表的数据
  43.         //如果添加，则需要按照 数据库名.表名 的格式指定，多个表使用逗号隔开
  44. //        .tableList("gmall-210325-flink.base_trademark")
  45.         .deserializer(new CustomerDeseriallization())
  46.  
  47.         //监控的方式：
  48.       // 1. initial 初始化全表拷贝，然后再比较
  49.       // 2. earliest 不做初始化，只从当前的
  50.       // 3. latest  指定最新的
  51.       // 4. specificOffset 指定offset
  52.       // 3. timestamp 比指定的时间大的
  53.  
  54.         .startupOptions(StartupOptions.latest())
  55.         .build()
  56.  
  57.     val  dataStream = env.addSource(sourceFunction)
  58.  
  59.     //3. sink, 写入kafka
  60.     dataStream.print()
  61.     val sinkTopic = "ods_base_db"
  62.     dataStream.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic))
  63.     //4. 启动任务
  64.     env.execute("flinkCDC")
  65.  
  66.   }
  67.  
  68. }
FlinkCDC.scala

3. 测试项目

准备好kafka，mysql，可以在本地测试。

启动kafka消费者，topic是ods_base_db

在idea中启动flinkcdc程序。

打开mysql编辑器，表base_trademark中原始记录有12条如下：

 现在手工增加一条记录，编号为13  wang

 查看idea控制台显示添加消息如下：

 同时在Kafka消费者也看到一条记录如下，字段type为操作类型，c表示创建

再次在MySQL中做修改和删除操作，可以看到控制多了两条记录，操作类型分别为u和d，表示修改和删除操作。

到此flinkcdc的操作基本完成。

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

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

  

闽ICP备14008679号