Johnson Lin

Apache Paimon（孵化中）是一个流式数据湖平台，支持高速数据摄取、变化数据跟踪和高效的实时分析。 架构 如上面的架构图所示： 读/写：Paimon 支持以多种方式读/写数据和执行 OLAP 查询。 对于读取，它支持消费 历史快照（批处理模式）的数据， 从最新偏移量（流处理模式）读取数据，或者 以混合方式读取增量快照。 对于写入，它支持从数据库的变化日志（CDC）中进行流式同步，或从离线数据中批量插入/覆盖。 生态系统：除了 Apache Flink 之外，Paimon 还支持其他计算引擎的读取，如 Apache Hive、Apache Spark 和 Trino。 内部原理：Paimon 在文件系统/对象存储中存储列式文件，并使用 LSM 树结构来支持大量数据更新和高性能查询。 统一存储对于像 Apache Flink 这样的流引擎，通常有三种类型的连接器： 消息队列，如 Apache Kafka，用于管道的源头和中间阶段，以确保延迟在秒级内。 OLAP 系统，如 Clickhouse，以流的方式接收处理后的数据，并为用户的临时查询提供服务。 批量存储，如...

本文介绍如何在 Flink 中使用 Paimon。 准备 Paimon Jar 文件Paimon 目前支持 Flink 1.17、1.16、1.15 和 1.14。推荐使用最新的 Flink 版本以获得更好的体验。 下载对应版本的 jar 文件： Flink 版本 Jar Flink 1.17 https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.17/0.4.0-incubating/paimon-flink-1.17-0.4.0-incubating.jar Flink 1.16 https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.16/0.4.0-incubating/paimon-flink-1.16-0.4.0-incubating.jar Flink...

Paimon 不仅原生支持 Flink SQL 的写入和查询，而且还提供其他流行引擎的查询，例如 Apache Spark 和 Apache Hive。 兼容性矩阵 引擎 版本 批读取 批写入 创建表 流写入 流读取 批覆盖 Flink 1.14 - 1.17 ✅ ✅ ✅ ✅ ✅ ✅ Hive 2.1 - 3.1 ✅ ✅ ❌ ❌ ❌ ❌ Spark 3.1 - 3.4 ✅ ✅ ✅ ❌ ❌ ❌ Spark 2.4 ✅ ❌ ❌ ❌ ❌ ❌ Trino 358 - 400 ✅ ❌ ❌ ❌ ❌ ❌ Presto 0.236 - 0.279 ✅ ❌ ❌ ❌ ❌ ❌ 正在进行的引擎： Doris：开发中，支持Paimon目录，Doris 产品路线图 2023。 Seatunnel：开发中，引入 Paimon 连接器。 Starrocks：正在讨论中。 （END）

除了底层表文件的变更日志外，Paimon 的变更日志也可以存储在外部日志系统中，比如 Kafka，或者从外部日志系统中消费。通过指定 log.system 表属性，用户可以选择使用哪个外部日志系统。 如果选择使用外部日志系统，那么所有写入表文件中的记录也会写入日志系统。因此，流查询产生的变化会从日志系统读取，而不是表文件。 一致性保证默认情况下，日志系统中的变化要等到快照之后才对消费者可见，就像表文件一样。这种行为保证了精确一次的语义。也就是说，每条记录只被消费者看到一次。 但是，用户也可以指定表的属性 'log.consistency' = 'eventual'，这样写进日志系统的变更日志就可以立即被消费者消费，而不用等待下一个快照。这种行为减少了变更日志的延迟，但由于可能发生的故障，它只能保证至少一次的语义（即，消费者可能会看到重复的记录）。 如果设置 'log.consistency' = 'eventual'，为了获得正确的结果，Flink 中的 Paimon...

如果一个表没有定义主键，那么它默认是一个 append-only 表。 你只能向表中插入完整的记录。不支持删除或更新，也不能定义主键。这种类型的表适用于不需要更新的用例（例如日志数据同步）。 分桶（Bucketing）你也可以为 Append-only 表定义桶的数量，见 Bucket。 建议设置 bucket-key 字段。否则，数据会按照整行进行散列，性能会很差。 压缩（Compaction）默认情况下，接收节点（sink 节点）将自动进行压缩以控制文件数量。以下选项控制压缩策略： Key 默认值 类型 描述 write-only false Boolean 如果设置为 true，压缩和快照过期将被跳过。此选项与专门的压缩作业一起使用。 compaction.min.file-num 5 Integer 对于文件集 [f_0,…,f_N]，满足 sum(size(f_i)) >= targetFileSize...

Changelog 表是创建表时的默认表类型。用户可以插入、更新或删除表中的记录。 主键由一列或多列组成，其值能唯一地标识表中的每一行记录。Paimon 强制对数据进行排序，这意味着系统将对每个桶内的主键进行排序。利用这个特性，用户可以通过在主键上添加过滤条件来实现高性能。 通过在变更日志表上定义主键，用户可以获得以下功能。 合并引擎当 Paimon 接收器（sink）收到具有相同主键的两条或更多记录时，它将把这些记录合并成一条记录以保持主键的唯一性。通过指定 merge-engine 表属性，用户可以选择如何合并记录。 在 Flink SQL TableConfig 中始终将 table.exec.sink.upsert-materialize 设置为NONE，sink upsert-materialize 可能会导致奇怪的行为。当输入乱序时，建议使用 Sequence Field 进行乱序校正。 去重（Deduplicate）deduplicate 合并引擎是默认的合并引擎。Paimon 只会保留最新的记录，并丢弃具有相同主键的其他记录。 具体来说，如果最新的记录是一条...

本文将深入探讨各种文件操作对文件产生的影响。文章提供具体的例子和实用技巧，说明 Paimon 是如何有效管理文件的。此外，通过对提交（commit）和压缩（compact）等操作的深入剖析，旨在加深对文件的创建和更新的理解。 先决条件在深入阅读本文之前，本文假设你已经了解了以下内容： Paimon 基本概念； Paimon 文件布局； 如何在 Flink 中使用 Paimon。 创建目录通过 ./sql-client.sh 启动 Flink SQL 客户端，并逐一执行以下语句来创建 Paimon 目录： 123456CREATE CATALOG paimon WITH ('type' = 'paimon','warehouse' = 'file:///tmp/paimon');USE CATALOG paimon; 执行该创建目录语句会在给定的路径 file:///tmp/paimon 创建一个目录： 12$ ls...

一个表的所有文件都存储在一个基本目录下。Paimon 文件是以分层的方式组织的。下图说明了文件的布局。从一个快照文件开始，Paimon 读取器可以递归地访问表中的所有记录。 快照文件（Snapshot Files）所有的快照文件都存储在 snapshot 目录中。 一个快照文件是一个 JSON 文件，包含有关此快照的信息，包括： 使用的模式文件 包含此快照中所有更改的清单列表 清单文件（Manifest Files）所有清单列表和清单文件都存储在 manifest 目录中。 清单列表是清单文件名称的列表。 清单文件是一个包含有关 LSM 数据文件和变更日志文件的变更文件。例如，在相应的快照中，哪个 LSM 数据文件被创建，哪个文件被删除。 数据文件（Data Files）数据文件是按分区和桶组织的。每个桶目录包含一个 LSM 树及其变更日志文件。 目前，Paimon 支持使用 orc（默认）、parquet 和 avro 作为数据文件的格式。 LSM 树（LSM Trees）Paimon 采用 LSM 树（log-structured...

快照（Snapshot）快照捕获了一个表在某个时间点上的状态。用户可以通过最新的快照来访问一个表的最新数据。通过时间旅行，用户也可以通过较早的快照访问表的先前状态。 分区（Partition）Paimon 采用了与 Apache Hive 相同的分区概念来分离数据。 分区是一种可选的方式，可以根据特定列的值（如日期、城市和部门等）将表划分为相关部分。每个表可以有一个或多个分区键来标识一个特定的分区。 通过分区，用户可以高效地对表中的某一记录切片进行操作。有关如何将文件划分为多个分区的详细信息，请参见文件布局。 如果定义了主键，分区键必须是主键的一个子集。 桶（Bucket）非分区的表或分区表中的分区会被细分为桶，以便为数据提供额外的结构，这可用于更有效的查询。 一个桶的范围是由记录中的一个或多个列的哈希值决定的。用户可以通过提供 bucket-key 选项来指定桶的列。如果没有指定 bucket-key...

语法1PFSELFTEST 可用版本 ≥ 2.8.9 时间复杂度 N/A ACL类别 @hyperloglog, @admin, @slow, @dangerous PFSELFTEST 是一个内部命令，仅用于开发和测试 Redis。 （END）

语法1PFMERGE destkey [sourcekey [sourcekey ...]] 可用版本 ≥ 2.8.9 时间复杂度 $O(N)$ N 为需要的 HyperLogLog 的数量。 ACL类别 @write, @hyperloglog, @slow 将多个 HyperLogLog 合并为一个 HyperLogLog，合并后的 HyperLogLog 的基数估算值是通过对所有给定 HyperLogLog 进行并集计算得出的。 合并后的 HyperLogLog 会被储存在 destkey 键里面，如果该键并不存在，那么命令在执行之前，会先为该键创建一个空的 HyperLogLog。如果该键存在，则将其视为源集之一，其基数将包含在计算的 HyperLogLog 的基数中。 简单来说就是，把多个 HyperLogLog 合并到一起，得到一个代表并集基数的HyperLogLog，如果目标 key 存在，也会一起合并，得到的目标 key 就是最终的合并结果。 返回值返回 OK。 示例 1将两个 HyperLogLog 合并为一个...

语法1PFDEBUG subcommand key 可用版本 ≥ 2.8.9 时间复杂度 N/A ACL类别 @write, @hyperloglog, @admin, @slow, @dangerous PFDEBUG 命令是一个内部命令，用于开发和测试 Redis。 （END）

语法1PFCOUNT key [key ...] 可用版本 ≥ 2.8.9 时间复杂度 当命令作用于单个 HyperLogLog 时，复杂度为 $O(1)$，并且具有非常低的平均常数时间。当命令作用于 N 个 HyperLogLog 时，复杂度为 $O(N)$ ，常数时间也比处理单个 HyperLogLog 时要大得多。 ACL类别 @read, @hyperloglog, @slow 调用 PFCOUNT 命令时，如果只传入一个 key，则返回存储在指定 key 中的 HyperLogLog 数据结构计算的近似基数，如果 key 不存在则返回为 0。 如果传入多个 key 时，则返回所有给定 key 中的 HyperLogLog 的合集的近似基数，这个近似基数是通过合并所有给定的 HyperLogLog 到一个临时 HyperLogLog 中计算得出的。 HyperLogLog 数据结构用于计算集合中的唯一元素，只需要使用少量固定内存，特别是每个 HyperLogLog 占用 12KB（加上 key 本身的几个字节）。 返回的集合基数不是精确的，而是近似的，标准误差为...

语法1PFADD key [element [element ...]] 可用版本 ≥ 2.8.9 时间复杂度 每添加一个元素的复杂度为 $O(1)$ ACL类别 @write, @hyperloglog, @fast 将所有的元素参数添加到 HyperLogLog 数据结构中，存储在作为第一个参数指定的变量名称中。 作为这个命令的副作用，在执行此命令时，HyperLogLog 内部可能会更新，以反映截至目前为止添加的唯一项目数的不同近似值（集合的基数）。 如果 HyperLogLog 估计的近似基数在执行命令后发生变化，PFADD 将返回 1，否则返回 0。如果指定的键不存在，该命令将自动创建一个空的 HyperLogLog 结构（即具有指定长度和给定编码的 Redis 字符串）。 在调用命令时，如果不传递元素，而只有有效的 key 名称，那么如果 key 已经存在则不执行任何操作，否则只创建数据结构（在后一种情况下会返回 1）。 有关 HyperLogLog 数据结构的更多信息，请参阅 PFCOUNT 命令文档。 返回值返回一个整数，如果 HyperLogLog...

语法1BITCOUNT key [start end [BYTE | BIT]] 可用版本 ≥ 2.6.0 时间复杂度 $O(N)$ ACL类别 @read, @bitmap, @slow 计算给定字符串中，被设置为 1 的比特位的数量。 默认情况下，给定的整个字符串都会被进行计数，通过指定额外的 start 和 end 参数，可以让计数只在特定的位上进行。 start 和 end 参数的设置和 GETRANGE key start end 命令类似，都可以使用负数值： 比如 -1 表示最后一个字节，-2 表示倒数第二个字节，以此类推。 不存在的 key 被当成是空字符串来处理，因此对一个不存在的 key 进行 BITCOUNT 操作，结果为 0 。 默认情况下，参数 start 和 end 指定一个字节索引。我们可以使用一个附加参数 BIT 来指定一个比特索引。所以 0 是第一位，1 是第二位，以此类推。对于负值，-1 是最后一位，-2 是倒数第二位，依此类推。 返回值返回一个整数，表示被设置为 1 的位的数量。 示例 1对一个不存在的 key 进行 BITCOUNT...

从 Redis 2.0.0 版本开始，此命令被标记为已废弃。在迁移或编写新的代码时，可以使用 GETRANGE 命令替代。 语法1SUBSTR key start end 可用版本 ≥ 1.0.0 时间复杂度 $O(N)$ 其中 N 是返回字符串的长度。复杂度最终是由返回的长度决定的，但是由于从现有的字符串中创建一个子串的代价是非常低的，对于小的字符串可以认为是 $O(1)$。 ACL类别 @read, @string, @slow 返回存储在 key 中的字符串值的子串，由偏移量 start 和 end 决定（两者均包括在内）。负的偏移量可以用来提供一个从字符串的末尾开始的偏移量。所以 -1 表示最后一个字符，-2 表示倒数第二个，以此类推。 该命令通过将结果范围限制为字符串的实际长度来处理超出范围的请求。 返回值返回存储在 key 中的字符串值的子串。 示例 112345redis> SET user "JOHNSON LIN"OKredis> SUBSTR user 1 2"OH" 示例...