bluishglc
发表于
2024-05-27 09:20:18

CDC 数据实时同步入湖的技术、架构和方案（截至2024年5月的现状调研）

最近，对“实时摄取 CDC 数据同步到数据湖”这一技术主题作了一系列深入的研究和验证，目前这部分工作已经告一段落，本文把截止目前（2024年5月）的研究结果和重要结论做一下梳理和汇总。为了能给出针对性的技术方案，我们必须收敛话题，对一些技术选型做了限制，在数据库这一侧，我们以 MySQL 作为示例进行演示（PG 等其他主流数据库理论上均可行），在数据湖这一侧，我们重点关注的是 Apache Hudi。 ...继续阅读 (55)

bluishglc
发表于
2024-05-22 14:14:24

Hudi 多表摄取工具 HoodieMultiTableStreamer 配置方法与示例

由于 Hudi 的 HoodieMultiTableStreamer / HoodieMultiTableDeltaStreamer 是一次处理多张 Hudi 表的写入，这些表既会有如 hoodie.deltastreamer.source.kafka.value.deserializer.class 这样相同的公共配置，也会有如 hoodie.datasource.write.recordkey.field 这样每张表每张表都不同的个性化配置，为此，HoodieMultiTableStreamer / H ...继续阅读 (88)

bluishglc
发表于
2024-05-20 15:25:42

CDC 数据入湖方案：Kafka Connect＞Kafka + Schema Registry＞Hudi ( HoodieMultiTableStreamer )

本方案的技术链路为：使用 Kafka Connect 配合 Debezium MySQL Source Connector 将 MySQL 的 CDC 数据（Avro 格式）接入到 Kafka ，然后通过 Hudi 的 HoodieMultiTableStreamer 将摄取的 CDC 数据写入到 Hudi 表中。整个链路由 Confluent Schema Registry 控制 Schema 的变更。本文和《CDC 数据入湖方案：Kafka Connect > Kafka + Schema Regis ...继续阅读 (41)

bluishglc
发表于
2024-05-19 10:15:00

CDC 数据入湖方案：Flink CDC＞Kafka + Schema Registry＞Hudi ( HoodieMultiTableStreamer )

本方案的技术链路为：使用 Flink CDC 将 MySQL 的 CDC 数据（Avro 格式）接入到 Kafka ，然后通过 Hudi 的 HoodieMultiTableStreamer 将摄取的 CDC 数据写入到 Hudi 表中。整个链路由 Confluent Schema Registry 控制 Schema 的变更。本文和《CDC 数据入湖方案：Flink CDC > Kafka + Schema Registry > Hudi ( Flink Connector ) 》介绍的方案类似，不同之 ...继续阅读 (64)

bluishglc
发表于
2024-05-18 10:15:00

使用 HoodieMultiTableStreamer 进行 Debezium CDC 多表同步入湖的研究报告

先介绍一下大的背景吧，我们已经能通过 Flink CDC 将整个数据库同步到 Kafka 中了，这一部分的实现方案已经汇总在了《Flink CDC 整库 / 多表同步至 Kafka 方案（附源码）》一文中。接下来要完成的是后半程的工作：读取 Kafka 的 Debezium CDC 数据写入到数据湖的 Hudi 表中，这样就能完整地实现整个数据库同步入湖的设计目标，当然，要求还是：“源库整库 / 多表 => Kafka”是一个作业，“Kakfa => Hudi 整库 / 多表”也是一个作业，这样才有比较强 ...继续阅读 (38)

bluishglc
发表于
2024-05-17 09:18:50

Flink 生态对 Confluent / Kafka Schema Registry 支持情况的研究报告

这几年，在流式链路上引入一个 Schema Registry 变得越来越流行，也越来越有必要， Schema Registry 能有效控制 Schema 的变更，合理推进 Schema Evolution，同时，引入它以后还能有效精简消息内容（特别是针对 Avro 格式），提升消息的传输效率，所以引入 Schema Registry 是有很多正向收益的。在 Flink 生态中，对 Confluent Schema Registry 的支持度如何呢？本文，我们来详细地梳理和总结一下。有关的组件主要是 Flin ...继续阅读 (98)

bluishglc
发表于
2024-05-16 09:38:06

Hudi HoodieStreamer 报错 DebeziumSource only support SchemaRegistryProvider 解决方法

在使用 HoodieStreamer / HoodieDeltaStreamer 从 Kafka 摄取 Debezium CDC 消息并自动解析写入到 Hudi 表时，我们可能会遇到这样一个问题：org.apache.hudi.utilities.sources.debezium.DebeziumSource only support org.apache.hudi.utilities.schema.SchemaRegistryProvider 这个问题本身的解决方法很简单，但是这个问题对整个 CDC 数据 ...继续阅读 (161)

bluishglc
发表于
2024-05-14 10:36:00

Flink CDC 的 Debezium Json 消息中文乱码的解决方法

在使用 Flink CDC 的 API 摄取 CDC 数据到 Kafka 的时候，如果使用的是 JsonDebeziumDeserializationSchema，那么会有很大概率遇到中文乱码问题，下面就是一个示例 ...继续阅读 (53)

bluishglc
发表于
2024-05-12 10:51:03

Hudi HoodieStreamer 报错 A column or function parameter with name ts_ms cannot be resolved 解决方法

在使用 HoodieStreamer 启动一个 CDC 数据实时入湖的作业中，遇到了这样一个报错：org.apache.spark.sql.AnalysisException:[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name ts_ms cannot be resolved. Did you mean one of the following? [after, op, before].; line ...继续阅读 (56)

bluishglc
发表于
2024-05-11 16:40:47

Flink：找不到 start-scala-shell.sh 或 flink-scala-shell 的解释

自 Flink 依赖的 Scala 版本提升到 2.12 后，就不再支持 Scala Shell （Scala REPL）了，这一点，在 [ 官方文档 ] 中有明确说明，具体是因为 [ FLINK-10911 ] 这个 Issue，但是该 Issue 已经关闭，且标记为：不会被解决，因为，根据社区投票，Flink 决定不再支持 Scala Shell （Scala REPL），详情可参见：[ FLINK-24360：Drop Scala Shell ]，所以，在这个问题上也就没有必要纠结了，老老实实编写 ...继续阅读 (65)

bluishglc
发表于
2024-05-11 15:02:23

Hudi 报错 NoSuchMethodException: MysqlDebeziumSource.＜init＞(...) 解决方法

在使用 Hudi 的 HoodieStreamer 或 HoodieMultiTableStreamer 接入 Debezium CDC 数据并写入 Hudi 表过程中，我们可能会遇到这样一个错误：NoSuchMethodException: org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource.(org.apache.hudi.common.config.TypedProperties, org.apache.spark.api. ...继续阅读 (59)

bluishglc
发表于
2024-04-28 10:24:42

Spark Structured Streaming 分流或双写多表 / 多数据源（Multi Sinks / Writes）

在 Spark Structured Streaming 中，我们有时候需要将最后的处理结果分流或双写到多张表或多个数据源（Multi Sinks / Writes），一个典型的例子是：在 CDC 数据入湖场景里，一个 Kafka Topic 上存放着整库或多张表的 CDC 消息，使用 Spark 从 Kafka 中摄取这些消息后，需要根据消息中提供的数据库名和数据表名对 CDC 消息分流，然后写到数据湖上对应的 ODS 表中，这就是一种典型的“数据分流”场景。在 Spark Structured Stre ...继续阅读 (80)

bluishglc
发表于
2024-04-27 09:43:25

Maven 构建 Flink 应用程序的最佳实践（根除各种类冲突/类加载问题）

作为开发者，在构建 Flink 应用程序的体验真是一言难尽，想必大家都曾遇到过各种 ClassNotFoundException、NoSuchMethodError 以及 Could not find any factory for identifier ‘kafka/jdbc/hive/hudi’ that implements org.apache.flink.table.factories.DynamicTableFactory in the classpath. 这样的错误。坦率地说，Flink 应 ...继续阅读 (43)

bluishglc
发表于
2024-04-26 11:02:54

Flink CDC / Kafka Connect 自动转换 Debezium 的 DataTime / Timpstamp 时间格式

不管是用 Flink CDC 还是 Kafka Connect (Debezium Connector)，在实时获取数据库的 CDC 数据并以 Json 格式写入 Kafak 中时，都会遇到 DataTime / Timpstamp 类型的转换问题，即：原始数据库中的 DataTime / Timpstamp 的字面量是 2021-12-14 00:00:00 这种形式，但是，转换为 Json 后就变成了 1639411200000 (带毫秒位的 epoch 时间)，这带来的问题是：下游表基于 Json 数 ...继续阅读 (62)

bluishglc
发表于
2024-04-25 09:59:27

Flink: Could not find any factory for identifier kafka/jdbc/hive implements DynamicTableFactory 根治方法

运行 Flink 应用或执行 Flink SQL 时会经常遇到下面的错误：org.apache.flink.table.api.ValidationException: Could not find any factory for identifier ‘???’ that implements ‘org.apache.flink.table.factories.DynamicTableFactory’ in the classpath.其中 identifier 可以是 kafka, jdbc, hive ...继续阅读 (55)

bluishglc
发表于
2024-04-24 11:11:54

Maven 安装第三方、非公共 Jar 包实现自动化构建

有时候，我们的程序可能需要使用到非官方的第三方 Jar 包，这些 Jar 包无法从 Public Maven Repo 中获得，但为了不影响自动化的构建流程，我们需要把这种 Jar 安装到构建服务器的本地 Maven Repository 中，以便整个项目的构建和部署依然能自动完成，而不必关系这些第三方 Jar 包如何部署与安置。本文我们就介绍一下具体操作。 ...继续阅读 (51)

bluishglc
发表于
2024-04-18 09:29:21

Flink CDC 整库 / 多表同步至 Kafka 方案（附源码）

本文讨论的方案将是本博客以往介绍的所有关于 CDC 数据同步方案中最贴合实际生产需要的，因为以往介绍的开箱即用方案往往都是一张表对应一个 Kafka Topic，在数据库和数据表数量很大的情况下，这种模式是不实用的，用户真正需要的是：将多张数据表或整个数据库写入到一个 Kafka Topic 中，然后下游再对 Kafka 消息进行分流，也就是把同一张表中的数据分离出来写入到下游表中。要实现这样的设计目标，有一个非常重要的前提条件：Kafka 中的消息必须要包含两个关键字段，就是这条 CDC 数据所属的数据 ...继续阅读 (56)

bluishglc
发表于
2024-04-17 13:27:02

Flink CDC：使用 Flink SQL 将多表写入一个 Kafka Topic 以及 Flink 作业数量的测试

本测试要验证两个问题：Flink CDC 能否将多张表的 CDC 数据（debezium-json 格式）写入到同一个 Kafka Topic 中？验证使用 Flink SQL 方式将多表同时写入 Kafka 时，Flink 的作业数量,首先，准备好用 Flink SQL 实现的将两张表同步到一个 Kafka Topic 中的代码 ...继续阅读 (51)

bluishglc
发表于
2024-04-17 10:52:39

Flink 报错 ClassNotFoundException: org.apache.kafka.connect.source.SourceRecord 解决方法

在使用 Flink CDC 的 API 编写整库、多表接入 Kafka 的程序时，遇到题目所述的报错。如果是自己编写的程序，可在 IDE 中直接查找该类是否在项目的依赖声明中；如果是部署的第三方应用，可以使用如下命令确认报错的类是否在所有依赖包中都不存在 ...继续阅读 (65)

bluishglc
发表于
2024-04-16 14:33:11

Flink CDC 的 debezium-json 格式和 debezium 原生格式是一回事吗？

这是一个很容易混淆和误解的问题，值得拿出来讨论对比一下。我们知道 Debezium 是专门用于捕获 CDC 数据的开源框架，它对接了多种数据库，同时也定义了自己的 CDC 数据交换格式，也就是常说的 `debezium` 格式。而Flink CDC 复用了 Debezium 的部分功能，也就是说：Debezium 是 Flink CDC 的底层采集工具，Flink CDC 的工程依赖会用使用到 Debezium 的 Jar 包，然后 Flink CDC 在 Debezium 基础之上，封装了额外的功能， ...继续阅读 (59)

bluishglc
发表于
2024-04-16 13:28:18

如何连通私有子网中的 MSK / Kafka 集群？

MSK 集群通常都是建在私有子网中的，这给本地访问带来了很多麻烦，特别是需要在本地使用 Kafka GUI 客户端管理和读写 MSK 数据的时候。本文会给出一套解决方案。我们这里讨论的问题有一点特殊性，那就是：由于 MSK 是托管服务，它的 Broker 主机名虽然映射了一个私有的 IP 地址，但是不能通过 SSH 登录的，这是比较特殊的一个地方，另一个地方是：Kafka 集群基本都是三个以上的 Broker，它的连接地址往往是三台以上的主机+端口号，这是另一个比较特殊的地方 ...继续阅读 (59)

bluishglc
发表于
2024-04-15 13:23:03

Flink SQL：debezium-json 格式的表一定是数据库的 CDC 数据吗？

debezium-json 格式有一种非常典型的应用场景，就是：上游（Source）是一张使用 Flink CDC 接入的关系数据库中的表，下游（Sink）是一张创建在 Kafka 上的表，这张表的 format 往往会定义为 debezium-json，以便 Flink 能获得全面的 CDC 信息用于流上的实时处理，这种场景我们在《Flink CDC 与 Kafka 集成：Snapshot 还是 Changelog？Upsert Kafka 还是 Kafka？》一文的 ”测试组合(1)：connect ...继续阅读 (55)

bluishglc
发表于
2024-04-15 11:13:31

Maven 项目 JDK 8、JDK 17 多版本 Java 编译依赖最佳实践

最近几年，整个 Java 生态圈正在经历并将长期出于从 JDK 8 到 JDK 17 或更高版本的升级换代中，较为典型是：以 Spring 为代表的 Web 应用开发框架大多已经升级到了 JDK 17，而在大数据生态圈，Flink、Spark 还在使用 JDK 8，对于那些多模块的 Maven 项目，会出现不同的 Module 使用不同版本的 JDK 问题，这给构建这类项目造成了一些困难，本文简单梳理一下这一问题的解决方法，给出最佳实践。 ...继续阅读 (51)

bluishglc
发表于
2024-03-23 09:03:35

Flink：“Lookup Join” 和 “基于处理时间的 Temporal Join” 的区别

在完整介绍了 “基于处理时间的 Temporal Join” 和 “Lookup Join” 之后，作为一个简单的盘点，也是回答最初接触这两种 Join 时的疑惑，我们来分析一下两种 Join 有何异同。相同之处：时效性都较高；时间维度上的关联准确性接近；不同之处：Lookup Join 是直连数据查询的，而 “基于处理时间的 Temporal Join” 是构建在 Flink 上的动态表，变更是靠 CDC 实时同步的。 ...继续阅读 (64)

bluishglc
发表于
2024-03-23 04:00:00

Flink：维表 Join 难点和技术方案汇总

目前看，Flink 的 “维表 Join” 主要就三种实现方式，叫法可能会有细微差别，以下是我是用更直白的语言总结的称谓：直连外部数据库进行关联；将维表加载到内存中关联；基于维表变更日志的关联。这些 Join 方案具体会使用到 Flink 的 Lookup Join、Temporal Join 等相关机制，所以在研究维表 Join 方案前，应先补齐这部分的知识，具体可参考本文末给出的本博客相关系列文章。 ...继续阅读 (75)

bluishglc
发表于
2024-03-22 13:25:14

一文聊透 Flink 的 Temporal Join

Flink的Temporal Join以及Temporal Table （时态表）关系密切，在知识结构上有点混乱，可能是历史原因造成的。本文希望把所有与 Temporal Join 相关的概念进行一次统一梳理，用更规范的知识结构把 Flink 的 Temporal Join 介绍清楚，包括：以 Temporal Table DDL 方式实现基于事件时间的 Temporal Join;以 Temporal Table DDL 方式实现基于处理时间的 Temporal Join;以 Temporal Table ...继续阅读 (62)

bluishglc
发表于
2024-03-22 10:03:01

Flink Temporal Join 系列 (4)：用 Temporal Table Function 实现基于处理时间的关联

本文要演示的是：使用 Temporal Table Function 定义被关联表（维表），然后基于主动关联表（事实表）的“处理时间”去进行 Temporal Join（关联时间维度上对应版本的维表数据）。由于 Flink 在新版本中已经禁止使用 Temporal Table DDL 实现基于处理时间的 Temporal Join，所以通过 Temporal Table Function 实现基于处理时间的 Temporal Join 已经是唯一的手段了。该演示同样涉及三个要点：被关联的表（维表）是用 Te ...继续阅读 (64)

bluishglc
发表于
2024-03-21 14:07:27

Flink：Lookup Join 实现与示例代码

本文要演示的是：在流上关联一张外部表（例如 MySQL 数据库中的一张维表），用于丰富流上的数据，实际上，这正是最普遍的 ”维表 Join“ 的实现方式。通过这种方式和外部维表关联时，依然能关联到最新变化的维度数据，所以才说这是 ”维表 Join“。Lookup Join 与《Flink Temporal Join 示例演示 (2)：Temporal Table DDL + 处理时间》一文演示情形是很相似的，但不同之处在于：这里的维表是通过 JDBC 方式连接的，而文章中的 Temporal Join 方 ...继续阅读 (62)

bluishglc
发表于
2024-03-21 09:36:08

Flink Temporal Join 系列 (3)：用 Temporal Table Function 实现基于事件时间的关联

本文要演示的是：使用 Temporal Table Function 定义被关联表（维表），然后基于主动关联表（事实表）的“事件时间”去进行Temporal Join（关联时间维度上对应版本的维表数据）。该演示同样涉及三个要点：被关联的表（维表）是用 Temporal Table Function 形式定义的时态表;主动关联的表（事实表）需要定义：“事件时间”属性（但并不需要是一张版本表）;Temporal Join 是使用 Temporal Table Function + LATERAL TABLE 关 ...继续阅读 (146)

bluishglc
发表于
2024-03-20 13:26:54

使用 Flink + Faker Connector 生成测试数据压测 MySQL

使用 Flink 压测 MySQL 是一个不错的注意，或者，有时候我们需要在 MySQL 中生成一些可控的测试数据，这时使用 Flink 的 Faker Connector 就是会很简单。本文记录一下操作方法。 ...继续阅读 (63)