[Тест Flink] Flink CDC DataStream API отслеживает динамическую отправку MySQL в Kafka Topic

Большие данные

1. Предпосылки

Предыстория бизнеса: добавочные данные MySQL обновляются в режиме реального времени и синхронизируются с Kafka для дальнейшего использования.

Проверьте Flink CDCофициальная документация, в котором в описании функций упоминаются различные уровни поддержки API SQL и DataStream.

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 удобен в использовании, он также прост. Однако из-за большого количества бизнес-таблиц, если для запуска Flink Job используется мониторинг одной таблицы, это принесет много хлопот по потреблению ресурсов и операциям по эксплуатации и обслуживанию, поэтому автор решил использовать DataStream API для реализовать однозадачный мониторинг MySQL CDC на уровне базы данных и отправлять данные в разные темы Kafka на основе имен таблиц.

Во-вторых, реализация кода

1. Ключевые зависимости 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

Flink CDC определяетcom.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchemaИнтерфейс используется для десериализации данных CDC. Класс реализации по умолчаниюcom.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchemaиcom.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema, Поскольку нам нужно настроить схему, мы не используем класс реализации по умолчанию в течение последних двух недель и реализуем интерфейс для определения нужной нам схемы.

определениеJsonDebeziumDeserializeSchemaвыполнить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 для мониторинга изменений данных базы данных MySQL:

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. Динамически отправлять данные в разные темы в Kafka

Из приведенной выше настроенной схемы мы можем узнать, что состав исходного поляmysql_binlog_source.库名.表名. На данный момент мы можем настроитьKafkaSerializationSchemaЧтобы добиться отправки разных данных в разные темы, то есть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))
    }
  }

В то же время определите метод создания производителей кафки, которые динамически отправляют данные в разные темы.

/**
   * 创建将数据动态发往不同topic的kafka生产者
   *
   * @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. Полная реализация основного класса

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()
  }
}

После запуска задачи видно, что какфа создала разные темы по названиям таблиц, и сохранила данные в разных таблицах.

На данный момент реализована функция использования DataStream API для мониторинга MySQL CDC на уровне базы данных в одной задаче и отправки данных в разные топики Kafka в соответствии с именем таблицы.

3. Резюме

В этой статье в основном представлена ​​функция мониторинга данных базы данных MySQL, отправляемых в разные темы в kafka через Flink CDC DataStream API, который используется для настройкиDebeziumDeserializationSchemaРеализовать синтаксический анализ и настройку пользовательской десериализации схемы CDC.KafkaSerializationSchemaРеализуйте такие функции, как отправка сообщений в разные темы на основе содержимого данных.

Категории