flink cdc实践

太过爱你忘了你带给我的痛 2022-09-15 11:52 256阅读 0赞

> flink 1.12.1  
> flink-cdc 1.2.0  
> mysql 7+

最近使用公司的大数据平台做实时ETL，发现平台提供的实时消费binlog到kafka的服务有点坑，莫名其妙失败重启，日志也没有效的信息，于是打算另辟蹊径。主要考虑能控制整个流程，能查看完整日志，不新增技术栈，最好还是基于flink，于是选择了flink-cdc，它底层基于debezium。

*  sterp 1  
    首先需要确保MySQL开启binlog，并且有采集binlog的全限

#需要的mysql配置
    log-bin = on #开启BINLOG
    binlog-format = ROW #选择ROW模式
    gtid_mode = ON #开启GTID
    
    #可以使用数据库用户执行以下命令确认是否有采集binlog全限
    #1、是否打开binlog特性，on表示开启
    show global variables where Variable_name = 'log_bin'
    #2、是否为row模式，row表示已开启row模式
    show global variables where Variable_name = 'binlog_format'
    #3、是否打开gtid特性，on表示开启
    show global variables where Variable_name = 'gtid_mode'
    #4、查询是否有获取binlog日志全限，为空则表示全选不够
    show binary logs #获取binlog文件列表
    show master logs #查看master的binlog日志
    show master status #查看当前写入的binlog文件
    show binlog events in 'mysql-bin.000001' #查看指定binlog文件内容
    
    #如果全限不够，可以参看下面的语句授权:
     GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'

有时候会报需要RELOAD权限，这时有两种办法：  
1、给用户赋予reload权限  
2、编写flinksql时配置debezium.snapshot.locking.mode=node参数，跳过全局锁

*  step2  
    依赖

<dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>flink-connector-mysql-cdc</artifactId>
      <version>1.2.0</version>
    </dependency>
    
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.18</version>
    </dependency>

demo

-- register a MySQL table 'orders' in Flink SQL
    CREATE TABLE orders (
      order_id INT,
      order_date TIMESTAMP(0),
      customer_name STRING,
      price DECIMAL(10, 5),
      product_id INT,
      order_status BOOLEAN
    ) WITH (
      'connector' = 'mysql-cdc',
      'hostname' = 'localhost',
      'port' = '3306',
      'username' = 'root',
      'password' = '123456',
      'database-name' = 'mydb',
      'table-name' = 'orders'
    );
    
    -- read snapshot and binlogs from orders table
    SELECT * FROM orders;

这样就能采集binlog日志了，下来就可以将采集的数据写入到下游，比如kafka，es，hudi等。

> 参考:  
> [https://ververica.github.io/flink-cdc-connectors/release-1.4/index.html][https_ververica.github.io_flink-cdc-connectors_release-1.4_index.html]  
> [https://nightlies.apache.org/flink/flink-docs-release-1.12/][https_nightlies.apache.org_flink_flink-docs-release-1.12]

[https_ververica.github.io_flink-cdc-connectors_release-1.4_index.html]: https://ververica.github.io/flink-cdc-connectors/release-1.4/index.html
[https_nightlies.apache.org_flink_flink-docs-release-1.12]: https://nightlies.apache.org/flink/flink-docs-release-1.12/