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

clougence 2023-03-28 原文

简述

实时数据处理领域中，使用 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	FlinkCDC	Maxwell
产品化	完备	基础	无
同步对象配置	可视化	代码	配置文件
封装格式	多种常用格式	自定义	JSON
高可用	有	有	无
数据初始化（snapshot）	实例级	实例级	单表
源端支持	ORACLE,MySQL,SQLServer,MongoDB,PostgreSQL...	ORACLE,MySQL,SQLServer,MongoDB,PostgreSQL...	MySQL

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 数据迁移同步。各位读者朋友，如果你觉得还不错，请点赞、评论加转发吧。

加入CloudCanal粉丝群掌握一手消息和获取更多福利，请添加我们小助手微信：suhuayue001

CloudCanal-免费好用的企业级数据同步工具，欢迎品鉴。
了解更多产品可以查看官方网站： http://www.clougence.com
CloudCanal社区：https://www.askcug.com/

有关5分钟搞定关系型数据库到 Flink 数据同步的更多相关文章

ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
ruby - Ruby 有 `Pair` 数据类型吗？ - 2
有时我需要处理键/值数据。我不喜欢使用数组，因为它们在大小上没有限制(很容易不小心添加超过2个项目，而且您最终需要稍后验证大小)。此外，0和1的索引变成了魔数(MagicNumber)，并且在传达含义方面做得很差(“当我说0时，我的意思是head...”)。散列也不合适，因为可能会不小心添加额外的条目。我写了下面的类来解决这个问题:classPairattr_accessor:head,:taildefinitialize(h,t)@head,@tail=h,tendend它工作得很好并且解决了问题，但我很想知道:Ruby标准库是否已经带有这样一个类？最佳
ruby - Rails 关联 - 同一个类的多个 has_one 关系 - 2
我的问题的一个例子是体育游戏。一场体育比赛有两支球队，一支主队和一支客队。我的事件记录模型如下:classTeam"Team"has_one:away_team,:class_name=>"Team"end我希望能够通过游戏访问一个团队，例如:Game.find(1).home_team但我收到一个单元化常量错误:Game::team。谁能告诉我我做错了什么？谢谢，最佳答案如果Gamehas_one:team那么Rails假设您的teams表有一个game_id列。不过，您想要的是games表有一个team_id列，在这种情况下
ruby - 我如何添加二进制数据来遏制 POST - 2
我正在尝试使用Curbgem执行以下POST以解析云curl-XPOST\-H"X-Parse-Application-Id:PARSE_APP_ID"\-H"X-Parse-REST-API-Key:PARSE_API_KEY"\-H"Content-Type:image/jpeg"\--data-binary'@myPicture.jpg'\https://api.parse.com/1/files/pic.jpg用这个:curl=Curl::Easy.new("https://api.parse.com/1/files/lion.jpg")curl.multipart_form_
世界前沿3D开发引擎HOOPS全面讲解——集3D数据读取、3D图形渲染、3D数据发布于一体的全新3D应用开发工具 - 2
无论您是想搭建桌面端、WEB端或者移动端APP应用，HOOPSPlatform组件都可以为您提供弹性的3D集成架构，同时，由工业领域3D技术专家组成的HOOPS技术团队也能为您提供技术支持服务。如果您的客户期望有一种在多个平台（桌面/WEB/APP，而且某些客户端是“瘦”客户端）快速、方便地将数据接入到3D应用系统的解决方案，并且当访问数据时，在各个平台上的性能和用户体验保持一致，HOOPSPlatform将帮助您完成。利用HOOPSPlatform，您可以开发在任何环境下的3D基础应用架构。HOOPSPlatform可以帮您打造3D创新型产品，HOOPSSDK包含的技术有：快速且准确的CAD
FOHEART H1数据手套驱动Optitrack光学动捕双手运动(Unity3D) - 2
本教程将在Unity3D中混合Optitrack与数据手套的数据流，在人体运动的基础上，添加双手手指部分的运动。双手手背的角度仍由Optitrack提供，数据手套提供双手手指的角度。 01 客户端软件分别安装MotiveBody与MotionVenus并校准人体与数据手套。MotiveBodyMotionVenus数据手套使用、校准流程参照：https://gitee.com/foheart_1/foheart-h1-data-summary.git02 数据转发打开MotiveBody软件的Streaming，开始向Unity3D广播数据；MotionVenus中设置->选项选择Unit
使用canal同步MySQL数据到ES - 2
文章目录一、概述简介原理模块二、配置Mysql使用版本环境要求1.操作系统2.mysql要求三、配置canal-server离线下载在线下载上传解压修改配置单机配置集群配置分库分表配置1.修改全局配置2.实例配置垂直分库水平分库3.修改group-instance.xml4.启动监听四、配置canal-adapter1修改启动配置2配置映射文件3启动ES数据同步查询所有订阅同步数据同步开关启动4.验证五、配置canal-admin一、概述简介canal是Alibaba旗下的一款开源项目，Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。Git地址：https://github.co
ruby-on-rails - 创建 ruby 数据库时惰性符号绑定(bind)失败 - 2
我正在尝试在Rails上安装ruby，到目前为止一切都已安装，但是当我尝试使用rakedb:create创建数据库时，我收到一个奇怪的错误:dyld:lazysymbolbindingfailed:Symbolnotfound:_mysql_get_client_infoReferencedfrom:/Library/Ruby/Gems/1.8/gems/mysql2-0.3.11/lib/mysql2/mysql2.bundleExpectedin:flatnamespacedyld:Symbolnotfound:_mysql_get_client_infoReferencedf
STM32读取串口传感器数据（颗粒物传感器，主动上传） - 2
文章目录1.开发板选择*用到的资源2.串口通信（个人理解）3.代码分析（注释比较详细）1.主函数2.串口1配置3.串口2配置以及中断函数4.注意问题5.源码链接1.开发板选择我用的是STM32F103RCT6的板子，不过代码大概在F103系列的板子上都可以运行，我试过在野火103的霸道板上也可以，主要看一下串口对应的引脚一不一样就行了，不一样的就更改一下。*用到的资源keil5软件这里用到了两个串口资源，采集数据一个，串口通信一个，板子对应引脚如下：串口1，TX：PA9，RX：PA10串口2，TX：PA2，RX：PA32.串口通信（个人理解）我就从串口采集传感器数据这个过程说一下我自己的理解，
SPI接收数据异常问题总结 - 2
SPI接收数据左移一位问题目录SPI接收数据左移一位问题一、问题描述二、问题分析三、探究原理四、经验总结最近在工作在学习调试SPI的过程中遇到一个问题——接收数据整体向左移了一位（1bit）。SPI数据收发是数据交换，因此接收数据时从第二个字节开始才是有效数据，也就是数据整体向右移一个字节（1byte）。请教前辈之后也没有得到解决，通过在网上查阅前人经验终于解决问题，所以写一个避坑经验总结。实际背景：MCU与一款芯片使用spi通信，MCU作为主机，芯片作为从机。这款芯片采用的是它规定的六线SPI，多了两根线：RDY和INT，这样从机就可以主动请求主机给主机发送数据了。一、问题描述根据从机芯片手

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

简述

技术点