CDC 的各种方案小结

支持的 Flink 版本

  获取dml query配置
 properties.setProperty("include.query","true");   


  添加新表扫描
.scanNewlyAddedTableEnabled(true)  eanbel scan the newly added tables fature


// output the schema changes as well 开启表结构变更支持
.includeSchemaChanges(true)

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.xsy.wc.model.OperateSqlModel;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


import java.util.Properties;


/**
 * author: lkn Date: 2022/4/22 ProjectName: flinkbase Version: 1.0
 */
public class MysqlStream {


    public static StreamExecutionEnvironment prepareEnvronment(){
         获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         设置并行度
        env.setParallelism(1);
         开启检查点   Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,
         需要从Checkpoint或者Savepoint启动程序
        2.1 开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        2.2 设置超时时间为1分钟
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        2.3 指定从CK自动重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 2000L));
         本次CheckpointingMode模式 精确一次 即exactly-once
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         同一时间只允许进行一个检查点
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        2.4 设置任务关闭的时候保留最后一次CK数据
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        2.5 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
      env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop52:9820/flinkCDC/chk-17");
        //2.6 设置访问HDFS的用户名
        System.setProperty("HADOOP_USER_NAME", "flowreplay");
        return env;
    }


    public static DataStreamSource <OperateSqlModel> buildStreamSource(StreamExecutionEnvironment env){
        Properties properties = new Properties();
        // debezium 配置
        properties.setProperty("include.query","true");
        MySqlSource<OperateSqlModel> mySqlSource = MySqlSource.<OperateSqlModel>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .scanNewlyAddedTableEnabled(true) // eanbel scan the newly added tables fature
                .databaseList("xsy_flowreplay") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("xsy_flowreplay.gen_table_copy1") // set captured table
//                .tableList("xsy_flowreplay.*") // set captured table
                .username("root")
                .password("root")
                .deserializer(new FlinkCdcDeserializationSchema()) // converts SourceRecord to JSON String
                .includeSchemaChanges(true)
                .debeziumProperties(properties)
                .build();
        // 读取数据封装流
        DataStreamSource <OperateSqlModel> mySQLDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source").setParallelism(1);
        mySQLDS.print(">>>").setParallelism(1);
        // 对流进行处理 包括过滤非法格式，并转换成json 字符串，发送给下游的kafka
        return mySQLDS;
    }


    /**
     * 执行命令
     *         bin/flink run \
     *         -t yarn-per-job \
     *         -d \
     *         -p 5 \
     *         -Drest.flamegraph.enabled=true \
     *         -Dyarn.application.queue=test \
     *         -Djobmanager.memory.process.size=1024mb \
     *         -Dtaskmanager.memory.process.size=2048mb \
     *         -Dtaskmanager.numberOfTaskSlots=2 \
     *         -Dmetrics.latency.interval=30000 \
     *         -c com.neo.flowreplay.data.sync.Mysqlcdc \
     * /opt/module/flink-1.13.1/myjar/FlowReplayDbSync-1.0-SNAPSHOT.jar
     * @param args
     */
    public static void main(String[] args) throws Exception {


        // 准备执行环境
        StreamExecutionEnvironment env=  prepareEnvronment();
        DataStreamSource<OperateSqlModel> sourceStream = buildStreamSource(env);
//        sourceStream.addSink(new MyMysqlSink()).setParallelism(1);
        env.execute("mysql test");
    }
}

import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.xsy.wc.model.OperateSqlModel;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;


/*
    flink 监听到的数据库变化数据的反序列化器
 */
public class FlinkCdcDeserializationSchema implements DebeziumDeserializationSchema <OperateSqlModel> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector <OperateSqlModel> collector) throws Exception {
        //  FlinkCDC采集数据格式
        try{
            Struct valueStruct = (Struct) sourceRecord.value();
            Struct sourceStrut = valueStruct.getStruct("source");
            //获取数据库的名称
            String database = sourceStrut.getString("db");
            //获取表名
            String table = sourceStrut.getString("table");
            // 获取完整sql
            String dml = sourceStrut.getString("query");


            //获取类型
            String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();


            //向下游传递数据
            collector.collect(new OperateSqlModel(type,dml));
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Override
    public TypeInformation <OperateSqlModel> getProducedType() {
        return TypeInformation.of(OperateSqlModel.class);
    }
}

DataStream<String> stream = ...


KafkaSink<String> sink = KafkaSink.<String>builder()
    // .setBootstrapServers(brokers)
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
    // 自定义 topic
    .setTopicSelector(new TopicSelector<Object>() {
      @Override
      public String apply(Object o) {
        // 此处写入逻辑即可
        return null;
      }
    })
    // 此处自定义分区
    .setPartitioner(new FlinkKafkaPartitioner<Object>() {
      @Override
      public int partition(Object o, byte[] bytes, byte[] bytes1, String s, int[] ints) {
        // 此处写入逻辑即可
        return 0;
      }
    })
    .setValueSerializationSchema(new SimpleStringSchema())
    .build()
)
.setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("t_0001")
.build();


stream.sinkTo(sink);

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "894683742",
      "rule-name": "894683742",
      "object-locator": {
        "schema-name": "market11",
        "table-name": "t_nft_token"
      },
      "rule-action": "include"
    },
    {
      "rule-type": "selection",
      "rule-id": "894683749",
      "rule-name": "894683749",
      "object-locator": {
        "schema-name": "assets",
        "table-name": "token_tx_flow"
      },
      "rule-action": "include"
    },
//  所有表中的modify_time改变类型为datetime
    {
      "rule-type": "transformation",
      "rule-id": "2",
      "rule-name": "2",
      "rule-action": "change-data-type", 


      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%",
        "column-name": "modify_time"
      },
      "data-type": {
        "type": "datetime"
      }
    },
 // 所有表中添加整型的hr，dt字段   
    {
      "rule-type": "transformation",
      "rule-id": "3",
      "rule-name": "3",
      "rule-action": "add-column",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "value": "hr",
      "expression": "strftime('%H',$create_time)",
      "data-type": {
        "type": "int4"
      }
    },
    {
      "rule-type": "transformation",
      "rule-id": "4",
      "rule-name": "4",
      "rule-action": "add-column",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "value": "dt",
      "expression": "date ($create_time)",
      "data-type": {
        "type": "date",
        "precision": 6
      }
    }
  ]
}