bluishglc
发表于
2024-03-20 09:49:35

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

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

bluishglc
发表于
2024-03-19 10:56:33

Flink：使用 Faker 和 DataGen 生成测试数据

DataGen 是开源 Flink 就内置的随机数据生成器；DataGen 生成的数据仅支持随机和序列值两种，且也并不是所有的数据类型都能支持随机或序列值，例如最见的一个需求：针对时间类型就不能生成指定区间内的单调递增的数值，相较而言，Faker 的功能要明显由于 DataGen，我们只需掌握 Faker 这一种数据生成器就足够了。 ...继续阅读 (100)

bluishglc
发表于
2024-03-19 08:57:41

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

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

bluishglc
发表于
2024-03-18 12:58:59

Linux：配置 rz 自动覆盖上传文件

Linux 上的 lz 和 rz 是两个很用的小工具。不少 Shell Client 工具都支持使用它们直接下载和上传文件。对 rz 来说，当一个文件已经存在服务器上时，再次上传同名文件时，为了避免覆盖掉现有文件，rz 选择给新上传的文件添加数字后缀，以作区分。这种处理方式虽然安全，但大多数情况下并不是用户希望的。 ...继续阅读 (160)

bluishglc
发表于
2024-03-18 09:26:47

再谈 Flink 的 “动态表” 和 “流表二象性”

本文再次去动态表和表流二象性做一些梳理。1.将流转换为动态表。2.在动态表上计算一个连续查询，生成一个新的动态表。3.生成的动态表被转换回流。我们会从实际代码层面把动态表的概念打通，核心问题就是：动态表的 DDL 定义了什么？持续查询又做了什么？动态表的 DDL 定义了三项核心要素：数据结构，数据源（connector），传输格式（format）；从 ETL 的角度看，持续查询完成了最核心的 ETL 逻辑，从整个流式处理管道的角度看，是持续查询驱动了整个 Pipeline 运转，只有动态表的DDL，不会有任 ...继续阅读 (64)

bluishglc
发表于
2024-03-03 12:57:38

Flink 定义 Temporal Table 的两种方式：Temporal Table DDL 和 Temporal Table Function

综合官方文档对 Temporal Joins 和 Temporal Table Function 两部分的描述可以梳理出以下重要结论：版本表的官方定义形式只有一种，就是 Temporal Table DDL，参考：声明版本表Temporal Table Function ( 时态表函数 ) 和 Temporal Table DDL 可以实现相同使用效果的时态表，可以说 Temporal Table Function ( 时态表函数 ) 相当于提供了 DDL 之外的另一种版本表的实现方式 ...继续阅读 (94)

bluishglc
发表于
2024-03-03 11:43:19

Flink：Temporal Table Function（时态表函数）和 Temporal Join

我们知道，时态表（确切地说应该是版本表）提供了回溯历史的能力，也就是能读取一条记录过去某个时刻所对应的值。要想查询版本表在过去某个时刻对应的值，我们得在查询时把这个时间作为参数传递给版本表，但这个时间参数绝不会是一个 where 条件，它是另一个维度（时间维度）上的参数，那么用怎样的形式才能把这个时间参数合理地表达到查询中呢？ Flink 使用了 UDF 的形式，主要思路就是：注册一个 UDF 来指代一张版本表，表名不能有参数，但函数可以有，这时把想访问版本表的目标时间点作为参数传给这个UDF，返回的就是当 ...继续阅读 (125)

bluishglc
发表于
2024-03-02 12:00:02

SQL 术语：Join 中的 Build 和 Probe 是什么意思？

我们可能在一些介绍数据库 Join 档中看到 Build 和 Probe，分别代表着 Join 操作中的右表和左表，为什么会有这样的称呼呢？原来它们都出自于一种叫 ”Hash Join“ 的 join 算法（常见的 Join 算法有：Hash Join、Loop Join、Merge Join）。先看一下名词解释：Hash Join：一种实现 Join 的算法，它通过在 Join 的一侧构建 Hash Table 并在另一侧不断匹配 Hash Table 来得到 Join 的结果。 ...继续阅读 (96)

bluishglc
发表于
2024-03-01 14:53:17

Flink：流式 Join 类型 / 分类盘点

在Flink中，实现流之间连接的操作可以分为两类。第一类是基于原生State状态存储的Connect算子操作，这种方式可以实现低延迟的数据连接和转换；第二类则是基于窗口的JOIN操作，这种方式又可以细分为window join和interval join两种，通过对数据进行时间窗口和滑动窗口的划分，实现不同粒度的数据关联和计算。Regular Join（常规 Join）Interval Join（时间区间 Join）Temporal Join (版本表 Join基于事件时间的 Temporal Join基于 ...继续阅读 (64)

bluishglc
发表于
2024-03-01 12:56:50

Flink：动态表 / 时态表 / 版本表 / 普通表概念区别澄清

根据 [ 官方文档 ] 所述，在 Flink 中，时态表和动态表是一个概念，只是强调的侧重点不同。Flink 流上的表都是动态的，也就是一直在变化，所以被称为动态表，因为动态表都会随时间发生变化，所以也被叫作了 “时态表”。而根据能否 trace (追踪) 一张时态表的变化历史，时态表会细分成：版本表和普通表两种，区别就是：版本表可以追溯历史，而普通表只保存当前最新状态的数据。Flink 官方文档中说：定义了主键约束和事件时间属性（通过 WATERMARK 关键字标识）的表就是版本表，并且举例说：数据 ...继续阅读 (109)

bluishglc
发表于
2024-02-27 13:36:32

Flink CDC 提取记录变更时间作为事件时间和 Hudi 表的 precombine.field 以及1970-01-01 取值问题

CDC 数据中的记录变更时间标记着这条记录在数据库中执行对应操作（创建/更新/删除）的时间，可以说是天然的“事件时间”，特别是对于那些本身没有记录时间字段的表来说就更加合适了。Flink 官方文档也建议在使用 CDC 的情况下，优先使用 CDC 中的这个时间字段，这个时间更加精准。与此同时，在定义 Hudi 表时，precombine.field 也是一个非常重要的配置，显然 CDC 数据中的记录变更时间是最合适的，没有之一。 ...继续阅读 (85)

bluishglc
发表于
2024-02-27 10:45:42

Flink SQL 中的流式概念：状态算子

传统的关系模型和 SQL 最开始都是为了批式处理而设计的，当把一个关系型查询应用到流式处理上时，在实现和转换的过程中，会有很多和批处理场景非常不同的地方，典型的例子就是：为了实现 SQL 的某些语义，Flink 必须在流上维持状态，典型的代表就是：连接、聚合、去重这些操作，它们都是“状态算子”，本质原因还是因为：流处理的表是无界的，流式查询是持续不停的，所以在流上维持状态是必须的。 ...继续阅读 (71)

bluishglc
发表于
2024-02-24 12:56:40

Flink：流上的“确定性”（Determinism）

先明确一下什么叫“确定性”：对于一个“操作”来说，如果每次给它的“输入”不变，操作输出的“结果”也不变，那么这个操作就是“确定性“的。通常，我们认为批处理的操作都是确定的，比如针对一张 clicks 表，假如表中的数据没有变化，无论我们执行多少次 SELECT * FROM clicks 操作，它的结果始终不变。但是，批处理操作并不一定总是“确定性”的 ...继续阅读 (71)

bluishglc
发表于
2024-02-22 12:43:09

使用 JMeter 生成测试数据对 MySQL 进行压力测试

数据库压测最主要的工作是：如何能简单地生成压测数据（Dummy Data），还要能控制好数据的取值范围，写脚本太麻烦，图形化工具收费还不通用，所以，我们选择使用老牌的压力测试工具：Apache JMeter，除了它内置了丰富的随机值生成方法外，还因为它是一个统一的压测平台，操作和配置都和其他类型的压测一致。本文以 Debezium 官方提供的 MySQL Docker镜像中的 Inventory 数据库为示例介绍和演示压测步骤。 ...继续阅读 (52)

bluishglc
发表于
2024-02-22 10:41:15

快速构建 Debezium MySQL Example 数据库

Debezium 官方提供的 MySQL Example 数据库是一个很好的工具，它内置了一个名为 inventory 的数据库，且已经配置好了 binlog，在测试 CDC 方案时非常有用，同时，它又可以基于 Docker 快速部署，所以使用起来就更方便了。本文专门介绍一下这个数据库的使用的方法。本文操作预设本地已安装并配置好了 docker，如需先行安装 docker，请参考《Apache Kafka 基于 S3 的数据导出、导入、备份、还原、迁移方案》一文第4节提供的 Shell 脚本。 ...继续阅读 (101)

bluishglc
发表于
2024-02-21 10:51:03

问题：Spark SQL 读不到 Flink 写入 Hudi 表的新数据，打开新 Session 才可见

使用 Flink 向 Hudi 表中写入数据，使用 Spark SQL 的 Shell 查询 Hudi 表（使用的是 Hudi HMS Catalog 统一管理和同步 Hudi 表的元数据），结果在 Spark 中只能查询到打开 Shell 之前表中的数据，之后通过 Flink 写入的数据不可见，但重新打开一个新的 Spark SQL Shell，就可以看到了。 ...继续阅读 (145)

bluishglc
发表于
2024-02-21 08:57:27

工具网站 • 生成测试数据（Generate Dummy Data）

介绍一个能自动生成各种测试数据（包括 INSERT INTO 的 SQL 语句）的工具网站:https://generatedata.com/ 🢡 Quickstart 页面，可点选现成的字段, https://generatedata.com/generator 🢡 全自定义界面，自行设置字段名、类型、取值范围等该工具支持所有主流的数据格式，包括：CSV、JSON、MYSQL 的 INSERT INTO 语句以及主流编程语言的集合类型声明语句等。下图是定制数据的操作界面截图： ...继续阅读 (60)

bluishglc
发表于
2024-02-20 13:07:22

CDC 整合方案：MySQL ＞ Flink CDC + Schema Registry + Avro ＞ Kafka ＞ Hudi

本文是《CDC 整合方案：MySQL ＞ Flink CDC ＞ Kafka ＞ Hudi》的增强版，借助 Flink CDC 从源头数据库接入数据并写入到 Kafka 后再使用 Flink 将数据写入到 Hudi 表中，在此过程中，使用了 Avro 格式传输数据，提升了效率，同时全程由 Confluent Schema Registry 参与，对 Schema Evolution 进行管控，该方案是一个完整的端到端的解决方案，也是最接近实际应用的选型方案之一。 ...继续阅读 (108)

bluishglc
发表于
2024-02-20 11:27:20

CDC 整合方案：MySQL ＞ Flink CDC ＞ Kafka ＞ Hudi

我们用 Flink CDC 完成 CDC 数据入湖任务：借助 Flink CDC 从源头数据库接入数据并写入到 Kafka 中，然后再使用 Flink 将数据写入到 Hudi 表中，所以这是一个完整的端到端的解决方案。上一篇文章我们省略了搭建 Kafka Connect + Debezium MySQL Connector 采集 CDC 数据的环节，因为这部分操作很复杂，Flink CDC 在应用和架构上确实比 Kafka Connect + Debezium MySQL Connector 的组合简单得多 ...继续阅读 (48)

bluishglc
发表于
2024-02-19 13:24:13

Flink Catalog 解读与同步 Hudi 表元数据的最佳实践

在当前的大数据格局中，Spark / Hive / Flink 是最为主流的 ETL 或 Streaming 引擎，元数据方面，Hive Metastore 可以视为事实上的 Data Catalog 标准，而在数据湖存储格式上，又有 Hudi、Iceberg 这类新晋的框架，在这种复杂的格局下，用户希望能它们之间能相互打通，以便能根据应用场景灵活地选择技术栈，同时又不会出现技术上的“隔离”，一个非常典型的例子是：当我们选择了 Hudi 作为数据湖的统一存储格式后，我们希望不管是 Flink 还是 Spar ...继续阅读 (89)

bluishglc
发表于
2024-02-19 11:17:31

通过 Upsert Kafka 透彻理解动态表（Dynamic Table）

“动态表”的本质是什么？它本质上是一条流（Stream）以及在流上维持着的若干状态（State）！upsert-kafka connector 不同于 kafka connector 的主要地方就在于：通过 upsert-kafka 创建的表能自动同步源端数据表的变更，（Table 模式下）查询该表的结果总能和源表实时保持一致，对于初次接触 Flink 的开发者来说会感到“非常神奇”，因为 upsert-kafka 的数据表是建立在 Kafka 上的，Kafka 并不支持更新和删除操作，但 upsert-k ...继续阅读 (78)

bluishglc
发表于
2024-02-05 15:19:28

Flink CDC 与 Kafka 集成：State Snapshot 还是 Changelog？Kafka 还是 Upsert Kafka？

我们知道，尽管 Flink CDC 可以越过 Kafka，将关系型数据库中的数据表直接“映射”成数据湖上的一张表（例如 Hudi 等），但从整体架构上考虑，维护一个 Kafka 集群作为数据接入的统一管道是非常必要的，这会带来很多收益。在 Flink CDC 之前，以 Debezium + Kafka Connect 为代表的技术组合都是将数据库的CDC数据先接入到 Kafka 中，然后再由后续的组件解析和处理。 ...继续阅读 (74)

bluishglc
发表于
2024-02-05 14:44:40

Flink 动态表 (Dynamic Table) 解读

根据过去在流上维持状态的编程经验，我们可以深刻地体会到：Dynamic Table 的本质其实是基于 changelog 数据流维持的一个流上的状态（State）！动态表是 Flink 能以 SQL 驱动和操纵流式处理的基础，也是 Flink 实现 ”批流一体“ 的一项内在的技术支撑。简单地说，它的思想就是：将一个”流“抽象成一张”无界”的数据表，这样就可以在上面施加 SQL 操作了。静态的关系表和数据流有可以类比的地方，这是能将两者映射在一起的理论基础，同时，它们之间也有难以弥合的差异，所以在某些方面 ...继续阅读 (72)

bluishglc
发表于
2024-02-02 17:57:08

Shell 脚本：按日期排序和查找文件 / 删除指定时间范围内的文件

有时候，我们会向一个文件夹里添加或删除文件，随着文件越来越多，操作历史越来越长，最初文件夹的状态可能就记不清了，某些时候，我们可能想恢复到文件夹的初始状态，重新来过，这时候就只能依靠文件修改时间来查找和过滤文件了。一个典型的场景就是：为了运行各种组件，我们会往一个 lib 文件夹中添加各种 Jar 包，有时候，时间久了，Jar 包可能会出现版本冲突，于是我们想清理到近期安装的各种 Jar 包，从初始状态重新开始。是以下字母的一些组合，但是在不同系统上，允许的组合种类也是不一样的，总得来说， ...继续阅读 (81)

bluishglc
发表于
2024-02-02 15:31:20

Flink SQL Client 安装各类 Connector、组件的方法汇总（持续更新中....）

一般来说，在 Flink SQL Client 中使用各种 Connector 只需要该 Connector 及其依赖 Jar 包部署到 ${FLINK_HOME}/lib 下即可。但是对于某些特定的平台，如果 AWS EMR、Cloudera CDP 等产品会有所不同，主要是它们中的某些 Jar 包可能被改写过，例如和 Hive Metastore 的交互，AWS EMR 就有另外一套 Metatstore：Glue Data Catalog，所以接口也做了相应的，所以，简单的复制开源的 Jar 包可能会 ...继续阅读 (73)

bluishglc
发表于
2024-02-01 12:38:21

CDC 整合方案：Flink 集成 Confluent Schema Registry 读取 Debezium 消息写入 Hudi

需要提醒的是：Kafka Connect上必须安装Debezium MySQL Connector和Confluent Avro Converter两个插件，因此需要在官方镜像的基础上手动添加这两个插件；在上面的 SQL 中，开头部分的几个 SET 语句非常重要，它们用于设置 Checkpoint，如果没有这些设置，写入操作都不会提交，看到的状况就是：作业流一直运行，没有报错，但是 Hudi 表不会有任何数据，关于这一问题以及这些 SET 语句的解释，已在。 ...继续阅读 (77)

bluishglc
发表于
2024-02-01 12:30:06

Flink 流式读取 Debezium CDC 数据写入 Hudi 表无法处理 -D / Delete 消息

问题场景是：使用 Kafka Connect 的 Debezium MySQL Source Connector 将 MySQL 的 CDC 数据（Avro 格式）接入到 Kafka 之后，通过 Flink 读取并解析这些 CDC 数据，然后以流式方式写入到 Hudi 表中，测试中发现，INSERT 和 UPDATE 消息都能很好的处理，但是，-D 类型的 Delete 消息被忽略了，即使已经开启了 ‘changelog.enabled’ = ‘true’ ，既然无效。 ...继续阅读 (80)

bluishglc
发表于
2024-01-31 12:21:14

Flink 读取 Kafka 消息写入 Hudi 表无报错但没有写入任何记录的解决方法

本问题发生的场景是：使用 Kafka Connect 的 Debezium MySQL Source Connector 将 MySQL 的 CDC 数据（Confluent Avro 格式）接入到 Kafka 之后，通过 Flink 读取并解析这些 CDC 数据，然后写入到 Hudi 表中。在测试过程中发现：启动写入作业后，Hudi 表中迟迟没有数据写入，而 Flink 作业也没有报错。实际上，这应该是一比较常见的问题，我们的测试环境并没有特别的配置，大多数初次进行集成的开发者，大概率都会遇到这一问题， ...继续阅读 (74)

bluishglc
发表于
2024-01-31 10:02:04

Flink 集成和使用 Hive Metastore

想在 Flink 中使用 Hive Metastore 其实只需要将 Flink Hive Connector 以及 Hive Metastore 有关的 Jar 包部署到 ${FLINK_HOME}/lib 下即可，稍后我们会介绍一下具体做法。但是，如果是 AWS EMR，会有所不同，主要是 EMR 中的默写包是改写过的，特别是和 Hive Metastore 的交互要多留心，因为 EMR 上还有另外一套 Metatstore：Glue Data Catalog，所以，简单的复制开源的 Jar 包可能会有 ...继续阅读 (77)

bluishglc
发表于
2024-01-30 13:39:20

Flink 集成 Debezium Confluent Avro 报 AvroTypeException: Found ＜topic＞.Value, expecting union 错误解决方法

在使用 Flink 读取 Debezium Confluent Avro 格式的 CDC 数据时，遇到了上述错误。背景介绍和具体操作请参考《Flink 集成 Debezium Confluent Avro ( format=debezium-avro-confluent )》一文，该错误是在查询 Kafka Sink Table 时（也就是真正读取并解析 Avro CDC 数据时）发生的，错误信息 AvroTypeException: Found .Value, expecting union ...继续阅读 (60)