[toc]

一、背景

業(yè)務(wù)背景: MySQL增量數(shù)據(jù)實時更新同步到Kafka中供下游使用

查看了一下Flink CDC的官方文檔，其中Features的描述中提到了SQL和DataStream API不同的支持程度。

Features

1. Supports reading database snapshot and continues to read binlogs with exactly-once processing even failures happen.

2. CDC connectors for DataStream API, users can consume changes on multiple databases and tables in a single job without Debezium and Kafka deployed.

3. CDC connectors for Table/SQL API, users can use SQL DDL to create a CDC source to monitor changes on a single table.

雖然SQL API使用很絲滑，也很簡單。但是由于業(yè)務(wù)表較多，若是使用一個表的監(jiān)聽就開啟一個Flink Job，會對資源消耗和運維操作帶來很大的麻煩，所以筆者決定使用DataStream API實現(xiàn)單任務(wù)監(jiān)聽庫級的MySQL CDC并根據(jù)表名將數(shù)據(jù)發(fā)往不同的Kafka Topic中。

二、代碼實現(xiàn)

1. 關(guān)鍵maven依賴

                <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>1.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.4.0</version>
        </dependency>

2. 自定義CDC數(shù)據(jù)反序列化器

Flink CDC定義了com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema接口用以對CDC數(shù)據(jù)進行反序列化。默認(rèn)實現(xiàn)類com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema和com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema,由于我們需要自定義Schema，所以我們不采用這兩周默認(rèn)的實現(xiàn)類，自己實現(xiàn)該接口定義我們需要的Schema.

定義JsonDebeziumDeserializeSchema實現(xiàn)DebeziumDeserializationSchema接口方法

class JsonDebeziumDeserializeSchema extends DebeziumDeserializationSchema[String] {

  private final val log: Logger = LoggerFactory.getLogger(classOf[JsonDebeziumDeserializeSchema])

  override def deserialize(sourceRecord: SourceRecord, collector: Collector[String]): Unit = {
    val op = Envelope.operationFor(sourceRecord)
    val source = sourceRecord.topic()
    val value = sourceRecord.value().asInstanceOf[Struct]
    val valueSchema: Schema = sourceRecord.valueSchema()
    if (op != Operation.CREATE && op != Operation.READ) {
      if (op == Operation.DELETE) {
        val data = extractBeforeData(value, valueSchema)
        val record = new JSONObject()
          .fluentPut("source", source)
          .fluentPut("data", data)
          .fluentPut("op", RowKind.DELETE.shortString())
          .toJSONString
        collector.collect(record)
      } else {
        val beforeData = extractBeforeData(value, valueSchema)
        val beforeRecord = new JSONObject()
          .fluentPut("source", source)
          .fluentPut("data", beforeData)
          .fluentPut("op", RowKind.UPDATE_BEFORE.shortString())
          .toJSONString
        collector.collect(beforeRecord)

        val afterData = extractAfterData(value, valueSchema)
        val afterRecord = new JSONObject()
          .fluentPut("source", source)
          .fluentPut("data", afterData)
          .fluentPut("op", RowKind.UPDATE_AFTER.shortString())
          .toJSONString
        collector.collect(afterRecord)
      }
    } else {
      val data = extractAfterData(value, valueSchema)
      val record = new JSONObject()
        .fluentPut("source", source)
        .fluentPut("data", data)
        .fluentPut("op", RowKind.INSERT.shortString())
        .toJSONString
      collector.collect(record)
    }
  }

  override def getProducedType: TypeInformation[String] = BasicTypeInfo.STRING_TYPE_INFO
  ...
}

定義MySqlSource監(jiān)聽MySQL庫數(shù)據(jù)變化：

val properties = new Properties()
properties.setProperty("snapshotMode", snapshotMode)

val mysqlCdcSource = MySQLSource.builder[String]()
   .hostname(hostname)
   .port(port)
   .databaseList(database)
   .tableList(tableName)
   .username(username)
   .password(password)
   .deserializer(new JsonDebeziumDeserializeSchema)
   .debeziumProperties(properties)
   .serverId(serverId)
   .build()

3. 數(shù)據(jù)動態(tài)發(fā)往Kafka不同的Topic

由上面自定義的Schema我們可以知道，source字段的構(gòu)成為mysql_binlog_source.庫名.表名。此時我們可以自定義KafkaSerializationSchema來實現(xiàn)將不同的數(shù)據(jù)發(fā)往不同的topic，即OverridingTopicSchema:

abstract class OverridingTopicSchema extends KafkaSerializationSchema[String] {
    val topicPrefix: String

    val topicSuffix: String

    val topicKey: String

    override def serialize(element: String, timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
      val topic = if (element != null && element.contains(topicKey)) {
        val topicStr = JSON.parseObject(element).getString(topicKey).replaceAll("\\.", "_")
        topicPrefix.concat(topicStr).concat(topicSuffix)
      } else null
      new ProducerRecord[Array[Byte], Array[Byte]](topic, element.getBytes(StandardCharsets.UTF_8))
    }
  }

同時定義創(chuàng)建將數(shù)據(jù)動態(tài)發(fā)往不同topic的kafka生產(chǎn)者的方法

/**
   * 創(chuàng)建將數(shù)據(jù)動態(tài)發(fā)往不同topic的kafka生產(chǎn)者
   *
   * @param boostrapServers          kafka集群地址
   * @param kafkaSerializationSchema kafka序列器
   * @return
   */
def createDynamicFlinkProducer(boostrapServers: String, kafkaSerializationSchema:       KafkaSerializationSchema[String]): FlinkKafkaProducer[String] = {
    if (StringUtils.isEmpty(boostrapServers))
      throw new IllegalArgumentException("boostrapServers is necessary")
    val properties = initDefaultKafkaProducerConfig(boostrapServers)
    properties.put(ACKS_CONFIG, "all")

    new FlinkKafkaProducer[String](DEFAULT_TOPIC, kafkaSerializationSchema,
      properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE)
  }

4. 主類完整實現(xiàn)

object Cdc2KafkaByStream {

  def main(args: Array[String]): Unit = {
    val parameterTool = ParameterTool.fromArgs(args)
    //cdc config
    val hostname = parameterTool.get("hostname")
    val port = parameterTool.getInt("port", 3306)
    val username = parameterTool.get("username")
    val password = parameterTool.get("password")
    val database = parameterTool.get("database")
    val tableName = parameterTool.get("tableName")
    val serverId = parameterTool.getInt("serverId")
    val snapshotMode = parameterTool.get("snapshotMode", "initial")
    //kafka config
    val kafkaBrokers = parameterTool.get("kafkaBrokers")
    val kafkaTopicPrefix = parameterTool.get("kafkaTopicPrefix", "topic_")
    val kafkaTopicSuffix = parameterTool.get("kafkaTopicSuffix", "")
    val kafkaTopicKey = parameterTool.get("kafkaTopicKey", "source")

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    ExecutionEnvUtils.configStreamExecutionEnv(env, parameterTool)
    ExecutionEnvUtils.parameterPrint(parameterTool)

    val properties = new Properties()
    properties.setProperty("snapshotMode", snapshotMode)

    val mysqlCdcSource = MySQLSource.builder[String]()
      .hostname(hostname)
      .port(port)
      .databaseList(database)
      .tableList(tableName)
      .username(username)
      .password(password)
      .deserializer(new JsonDebeziumDeserializeSchema)
      .debeziumProperties(properties)
      .serverId(serverId)
      .build()

    val kafkaSink = KafkaUtils.createDynamicFlinkProducer(kafkaBrokers, new OverridingTopicSchema() {
      override val topicPrefix: String = kafkaTopicPrefix
      override val topicSuffix: String = kafkaTopicSuffix
      override val topicKey: String = kafkaTopicKey
    })
    env.addSource(mysqlCdcSource).addSink(kafkaSink).setParallelism(1)
    env.execute()
  }
}

啟動任務(wù)后可以看到kakfa中根據(jù)表名創(chuàng)建了不同的topic，并保存了不同表里的數(shù)據(jù)。

至此，實現(xiàn)了使用DataStream API單任務(wù)監(jiān)聽庫級的MySQL CDC并根據(jù)表名將數(shù)據(jù)發(fā)往不同的Kafka Topic的功能。

三、小結(jié)

本文主要介紹了通過Flink CDC DataStream API實現(xiàn)監(jiān)聽MySQL庫數(shù)據(jù)發(fā)往kafka不同Topic的功能，其中運用到自定義DebeziumDeserializationSchema實現(xiàn)CDC Schema自定義反序列化解析以及自定義KafkaSerializationSchema實現(xiàn)根據(jù)數(shù)據(jù)內(nèi)容將消息發(fā)送到不同的topic等功能。