关于数据库:5分钟搞定-关系型数据库-到-Flink-数据同步

简述

实时数据处理畛域中，应用 Flink 形式，除了从日志服务订阅埋点数据外，总离不开从关系型数据库订阅并解决相干业务数据，这时就须要监测并捕捉数据库增量数据，将变更按产生的程序写入到消息中间件以供计算（或生产）。
本文次要介绍如何通过 CloudCanal 疾速构建一条高效稳固运行的 MySQL -> Kafka -> Flink 数据同步链路。

技术点

兼容多种常见音讯构造

CloudCanal 目前反对 Debezium Envelope (新增)、Canal、Aliyun DTS Avro 等多种风行音讯构造，对数据上游生产比拟敌对。
本次对 Debezium Envelope 音讯格局的反对，咱们采纳了一种轻量的形式做到齐全兼容，充分利用 CloudCanal 增量组件，扩大数据序列化器 (EnvelopDeserialize)，失去 Envelop 音讯并发送到 Kafka 中。
其中 Envelop 的音讯构造分为 Payload 和 Schema 两局部

• Payload: 存储具体数据
• Schema: 定义 Payload 的解析格局 (默认敞开)

{
  "payload":{
    "after":{
      "column_1":"3",
      ...
    },
    "before":null,
    "op":"c",
    "source":{
      "db":"kafka_test",
      "table":"new_table"
      "pos":110341861,
      "ts_ms":1659614884026,
      ...
    },
    "ts_ms":1659614884026
  },
  "schema":{
    "fields":[
      {
        "field":"after",
        "fields":[
          {
            "field":"column_1",
            "isPK":true,
            "jdbType":4,
            "type":"int(11)"
          },
          ...
        ],
        "type":"struct"
      },
      ...
    ],
    "type":"struct"
  }
}

高度可视化的 CDC

CDC 工具如 FlinkCDC、Maxwell、Debezium … 各有特色，CloudCanal 绝对这些产品，最大的特点是高度可视化，自动化，下表针对指标端为 Kafka 的 CDC 简要做了一些比照。

CloudCanal 在均衡性能的根底上，提供多种关系型数据源的同步，以及反向同步；提供便捷的可视化操作、笨重的数据源增加、轻便的参数配置；
提供多种常见的音讯格局，仅仅通过鼠标点击，就能够应用其余 CDC 的音讯格局的传输，让数据处理变的异样的快捷、不便。
其中通过咱们在雷同环境的测试下，CloudCanal 在高写入的 MySQL 场景中，解决数据的效率体现的很杰出，后续咱们会持续对 CloudCanal 进行优化，晋升整体的性能。
综上，相比与相似的 CDC 产品来说，CloudCanal 简略笨重并集成一体化的操作占据了很大的劣势。

无缝对接 Flink 流式计算

Flink 流式计算中不仅要订阅日志服务器的日志埋点信息，同样须要业务数据库中的信息，通过 CDC 工具订阅数据，能缩小查问对业务数据库产生的压力还能以流的模式传输，不便与日志服务器中的数据进行关联解决。
理论开发中，能够将业务数据库中的信息提取过滤之后动静的放入 Hbase 中作为维度数据，不便相关联的宽表进行关联查问；
也能够对数据进行开窗、分组、聚合，同样也能够下沉到其余的 Kafka 消费者组中，实现数据的分层。

操作示例

前置条件

• 本例应用 Envelop 音讯格局，关系型数据库 MySQL 为示例，展现 MySQL 对接 Flink 的 Demo
• 登陆 CloudCanal SaaS 版，应用参见疾速上手文档
• 筹备好 1 个 MySQL 实例，1 个 Kafka 实例（本例应用本人搭建的 MySQL 5.6，阿里云 Kafka 2.2）
• 筹备好 Flink 生产端程序，配置好相干信息：flink-demo 下载
• 登录 CloudCanal 平台，增加 Kafka，MySQL

• Kafka 自定义一个主题 topic_1，并创立一条 MySQL -> Kafka 链路作为增量数据起源

工作创立

• 首先配置 FlinkDemo 程序的 阿里云 Kafka 相干信息

• 运行 FlinkDemo 程序，期待生产 MySQL 同步 Kafka 的数据（程序不要敞开）

• 工作治理 -> 工作创立
• 测试链接 并抉择 源 和 指标 数据库， 并抉择 DebeziumEnvelope 音讯格局，和 topic_1 主题 (在阿里云里提前创立)

• 抉择 数据同步 ，不勾选 全量数据初始化，其余选项默认

• 抉择须要迁徙同步的表 table1 和对应的 Kafka 主题 topic_1

继续点击下一步，并创立出数据同步工作。

Flink 生产数据

• 向 MySQL 生成数据，MySQL -> Kafka(topic_1) -> Flink
• FlinkDemo 接管到 Kafka(topic_1) 数据，下沉到 topic_2 主题，打印并输入；这里 Flink 程序 能够做更多的流式计算的操作 ，FlinkDemo 只是演示了 最根本的数据传输案例。

常见问题

还反对哪些源端数据源呢？

目前凋谢 MySQL、Oracle，SQLServer，Postgres，MongoDB 到 Kafka，如果各位有需要，能够在社区反馈给咱们。

反对 DDL 音讯同步吗?

目前 关系型数据到 kafka 是反对 DDL 音讯的同步的，能够将 关系型数据库 DDL 的变动同步到 Kafka 当中。

总结

本文简略介绍了如何应用 CloudCanal  进行 MySQL -> Kafka -> Flink 数据迁徙同步。各位读者敌人，如果你感觉还不错，请点赞、评论加转发吧。