聊聊流式数据湖Paimon(一)

翻译自 Apache Paimon官方文档

概览

概述

Apache Paimon (incubating) 是一项流式数据湖存储技术，可以为用户提供高吞吐、低延迟的数据摄入、流式订阅以及实时查询能力。

简单来说，Paimon的上游是各个CDC，即changlog数据流；而其自身支持实时sink与search(下沉与查询)changlog数据流。一般会与Flink等流式计算引擎集成使用。

流式数据湖是一种先进的数据存储架构，专门为处理大规模实时数据流而设计。在流式数据湖中，数据以流的形式持续不断地进入系统，而不是批量存储后处理。

数据湖是一个存储企业的各种各样原始数据的大型仓库，其中的数据可供存取、处理、分析及传输。

数据仓库中的数据是经过优化后(也可以看作是结构化的数据)，且与该数据仓库支持的数据模型吻合的数据。

Paimon提供以下核心功能：

• 统一批处理和流式处理：Paimon支持批量写入和批量读取，以及流式写入更改和流式读取表change log。
• 数据湖：Paimon作为数据湖存储，具有成本低、可靠性高、元数据可扩展等优点。
• Merge Engines：Paimon支持丰富的合并引擎（Merge Engines）。默认情况下，保留主键的最后一个条目。您还可以使用“部分更新”或“聚合”引擎。
• Changelog Producer：用于在数据湖中生成和跟踪数据的变更日志（changelog）；Paimon 支持丰富的 Changelog Producer，例如“lookup”和“full-compaction”；正确的changelog可以简化流式处理管道的构造。
• Append Only Tables：Paimon支持只追加(append only)表，自动压缩小文件，提供有序的流式读取。您可以使用它来替换消息队列。

架构

架构如下所示 读/写：Paimon 支持多种读/写数据和执行 OLAP 查询的方式。

• 对于读取，支持如下三种方式消费数据
  • 历史快照（批处理模式）
  • 最新的偏移量（流模式）
  • 混合模式下读取增量快照
• 对于写入，它支持来自数据库变更日志（CDC）的流式同步或来自离线数据的批量插入/覆盖。

生态系统：除了Apache Flink之外，Paimon还支持Apache Hive、Apache Spark、Trino等其他计算引擎的读取。 底层存储：Paimon 将列式文件存储在文件系统/对象存储上，并使用 LSM 树结构来支持大量数据更新和高性能查询。

统一存储

对于 Apache Flink 这样的流引擎，通常有三种类型的connector：

• 消息队列，例如 Apache Kafka，在该消息管道(pipeline)的源阶段和中间阶段使用，以保证延迟保持在秒级。
• OLAP系统，例如ClickHouse，它以流方式接收处理后的数据并服务用户的即席查询。
• 批量存储，例如Apache Hive，它支持传统批处理的各种操作，包括INSERT OVERWRITE。

Paimon 提供抽象概念的表。 它的使用方式与传统数据库没有什么区别：

• 在批处理执行模式下，它就像一个Hive表，支持Batch SQL的各种操作。 查询它以查看最新的快照。
• 在流执行模式下，它的作用就像一个消息队列。 查询它的行为就像从历史数据永不过期的消息队列中查询stream changelog。

基本概念

Snapshot

snapshot捕获table在某个时间点的状态。 用户可以通过最新的snapshot来访问表的最新数据。通过time travel，用户还可以通过较早的快照访问表的先前状态。

Partition

Paimon 采用与 Apache Hive 相同的分区概念来分离数据。 分区是一种可选方法，可根据date, city, and department等特定列的值将表划分为相关部分。每个表可以有一个或多个分区键来标识特定分区。 通过分区，用户可以高效地操作表中的一片记录。

Bucket

未分区表或分区表中的分区被细分为Bucket(桶)，以便为可用于更有效查询的数据提供额外的结构。 Bucket的范围由record中的一列或多列的哈希值确定。用户可以通过提供bucket-key选项来指定分桶列。如果未指定bucket-key选项，则主键（如果已定义）或完整记录将用作存储桶键。 Bucket是读写的最小存储单元，因此Bucket的数量限制了最大处理并行度。 不过这个数字不应该太大，因为它会导致大量 小文件和低读取性能。 一般来说，每个桶中建议的数据大小约为200MB - 1GB。

Consistency Guarantees

Paimon Writer 使用两阶段提交协议以原子方式将一批record提交到Table中。每次提交时最多生成两个snapshot。 对于任意两个同时修改table的写入者，只要他们不修改同一个Bucket，他们的提交就可以并行发生。如果他们修改同一个Bucket，则仅保证快照隔离。也就是说，最终表状态可能是两次提交的混合，但不会丢失任何更改。

文件

概述

一张表的所有文件都存储在一个基本目录下。 Paimon 文件以分层方式组织。 下图说明了文件布局。 从snapshot文件开始，Paimon reader可以递归地访问表中的所有记录。

Snapshot Files

所有snapshot文件都存储在snapshot目录中。 snapshot文件是一个 JSON 文件，包含有关此snapshot的信息，包括

• 正在使用的Schema文件
• 包含此snapshot的所有更改的清单列表(manifest list)

Manifest Files

所有清单(manifest)列表和清单文件都存储在清单目录中。 清单列表(manifest list)是清单文件名的列表。 清单文件是包含有关 LSM 数据文件和changelog文件的更改的文件。 例如对应快照中创建了哪个LSM数据文件、删除了哪个文件。

Data Files

数据文件按分区和桶(Bucket)分组。每个Bucket目录都包含一个 LSM 树及其changelog文件。 目前，Paimon 支持使用 orc（默认）、parquet 和 avro 作为数据文件格式。

LSM-Trees

Paimon 采用 LSM 树（日志结构合并树）作为文件存储的数据结构。 如下简要介绍

Sorted Runs

LSM 树将文件组织成多个 sorted runs。 sorted runs由一个或多个数据文件组成，并且每个数据文件恰好属于一个 sorted runs。 数据文件中的记录按其主键排序。 在 sorted runs中，数据文件的主键范围永远不会重叠。

如图所示的，不同的 sorted runs可能具有重叠的主键范围，甚至可能包含相同的主键。查询LSM树时，必须合并所有 sorted runs，并且必须根据用户指定的合并引擎和每条记录的时间戳来合并具有相同主键的所有记录。 写入LSM树的新记录将首先缓存在内存中。当内存缓冲区满时，内存中的所有记录将被顺序并刷新到磁盘，并创建一个新的 sorted runs。

Compaction

当越来越多的记录写入LSM树时，sorted runs的数量将会增加。由于查询LSM树需要将所有 sorted runs合并起来，太多 sorted runs将导致查询性能较差，甚至内存不足。 为了限制 sorted runs的数量，我们必须偶尔将多个 sorted runs合并为一个大的 sorted runs。 这个过程称为压缩。 然而，压缩是一个资源密集型过程，会消耗一定的CPU时间和磁盘IO，因此过于频繁的压缩可能会导致写入速度变慢。 这是查询和写入性能之间的权衡。 Paimon 目前采用了类似于 Rocksdb 通用压缩的压缩策略。 默认情况下，当Paimon将记录追加到LSM树时，它也会根据需要执行压缩。 用户还可以选择在专用压缩作业中执行所有压缩。

可以将 sorted runs 理解为多个有序的Data File组成的一个有序文件。

主键表

Changelog表是创建表时的默认表类型。用户可以在表中插入、更新或删除记录。 主键由一组列组成，这些列包含每个记录的唯一值。Paimon通过对每个bucket中的主键进行排序来实现数据排序，允许用户通过对主键应用过滤条件来实现高性能。 通过在变更日志表上定义主键，用户可以访问以下特性。

Bucket

桶（Bucket）是进行读写操作的最小存储单元，每个桶目录包含一个LSM树。

Fixed Bucket

配置一个大于0的桶，使用Fixed bucket模式，根据Math.abs(key_hashcode % numBuckets)来计算记录的桶。 重新缩放桶只能通过离线进程进行。桶的数量过多会导致小文件过多，桶的数量过少会导致写性能不佳。

Dynamic Bucket

配置'Bucket'='-1'。 先到达的key会落入旧的bucket，新的key会落入新的bucket，bucket和key的分布取决于数据到达的顺序。 Paimon 维护一个索引来确定哪个键对应哪个桶。 Paimon会自动扩大桶的数量。

• Option1: 'dynamic-bucket.target-row-num'：控制一个桶的目标行数。
• Option2：'dynamic-bucket.initial-buckets'：控制初始化bucket的数量。

Normal Dynamic Bucket Mode

当更新不跨分区（没有分区，或者主键包含所有分区字段）时，动态桶模式使用 HASH 索引来维护从键到桶的映射，它比固定桶模式需要更多的内存。 如下：

• 一般来说，没有性能损失，但会有一些额外的内存消耗，一个分区中的 1 亿个条目多占用 1 GB 内存，不再活动的分区不占用内存。
• 对于更新率较低的表，建议使用此模式，以显着提高性能。

Cross Partitions Upsert Dynamic Bucket Mode

当需要跨分区upsert（主键不包含所有分区字段）时，Dynamic Bucket模式直接维护键到分区和桶的映射，使用本地磁盘，并在启动流写作业时通过读取表中所有现有键来初始化索引 。 不同的合并引擎有不同的行为：

1. Deduplicate：删除旧分区中的数据，并将新数据插入到新分区中。
2. PartialUpdate & Aggregation：将新数据插入旧分区。
3. FirstRow：如果有旧值，则忽略新数据。

性能：对于数据量较大的表，性能会有明显的损失。而且，初始化需要很长时间。 如果你的upsert不依赖太旧的数据，可以考虑配置索引TTL来减少索引和初始化时间： 'cross-partition-upsert.index-ttl'：rocksdb索引和初始化中的TTL，这样可以避免维护太多索引而导致性能越来越差。 但请注意，这也可能会导致数据重复。

Merge Engines

当Paimon sink收到两条或更多具有相同主键的记录时，它会将它们合并为一条记录以保持主键唯一。 通过指定merge-engine属性，用户可以选择如何将记录合并在一起。

Deduplicate

deduplicate合并引擎是默认的合并引擎。 Paimon 只会保留最新的记录，并丢弃其他具有相同主键的记录。 具体来说，如果最新的记录是DELETE记录，则所有具有相同主键的记录都将被删除。

Partial Update

通过指定 'merge-engine' = 'partial-update'，用户可以通过多次更新来更新记录的列，直到记录完成。 这是通过使用同一主键下的最新数据逐一更新值字段来实现的。 但是，在此过程中不会覆盖空值。 如下所示：

• <1, 23.0, 10, NULL>-
• <1, NULL, NULL, 'This is a book'>
• <1, 25.2, NULL, NULL>

假设第一列是主键key，那么最后的结果是 <1, 25.2, 10, 'This is a book'>

Sequence Group

Sequence字段并不能解决多流更新的部分更新表的乱序问题，因为多流更新时 Sequence(序列) 字段可能会被另一个流的最新数据覆盖。 因此我们引入了部分更新表的序列组（Sequence Group）机制。 它可以解决：

1. 多流更新时出现混乱。 每个流定义其自己的序列组。
2. 真正的部分更新，而不仅仅是非空更新。

如下所示：

CREATE TABLE T (
  k INT,
  a INT,
  b INT,
  g_1 INT,
  c INT,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine'='partial-update',
  'fields.g_1.sequence-group'='a,b',
  'fields.g_2.sequence-group'='c,d'
);

INSERT INTO T VALUES (1, 1, 1, 1, 1, 1, 1);

-- g_2 is null, c, d should not be updated
INSERT INTO T VALUES (1, 2, 2, 2, 2, 2, CAST(NULL AS INT));

SELECT * FROM T; -- output 1, 2, 2, 2, 1, 1, 1

-- g_1 is smaller, a, b should not be updated
INSERT INTO T VALUES (1, 3, 3, 1, 3, 3, 3);

SELECT * FROM T; -- output 1, 2, 2, 2, 3, 3, 3

对于 sequence-group，有效的比较数据类型包括：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ。

Aggregation

可以为输入字段指定聚合函数，支持聚合中的所有函数。 如下所示：

CREATE TABLE T (
          k INT,
          a INT,
          b INT,
          c INT,
          d INT,
          PRIMARY KEY (k) NOT ENFORCED
) WITH (
     'merge-engine'='partial-update',
     'fields.a.sequence-group' = 'b',
     'fields.b.aggregate-function' = 'first_value',
     'fields.c.sequence-group' = 'd',
     'fields.d.aggregate-function' = 'sum'
 );
INSERT INTO T VALUES (1, 1, 1, CAST(NULL AS INT), CAST(NULL AS INT));
INSERT INTO T VALUES (1, CAST(NULL AS INT), CAST(NULL AS INT), 1, 1);
INSERT INTO T VALUES (1, 2, 2, CAST(NULL AS INT), CAST(NULL AS INT));
INSERT INTO T VALUES (1, CAST(NULL AS INT), CAST(NULL AS INT), 2, 2);


SELECT * FROM T; -- output 1, 2, 1, 2, 3

Default Value

如果无法保证数据的顺序，仅通过覆盖空值的方式写入字段，则读表时未覆盖的字段将显示为空。

CREATE TABLE T (
                  k INT,
                  a INT,
                  b INT,
                  c INT,
                  PRIMARY KEY (k) NOT ENFORCED
) WITH (
     'merge-engine'='partial-update'
     );
INSERT INTO T VALUES (1, 1, CAST(NULL AS INT), CAST(NULL AS INT));
INSERT INTO T VALUES (1, CAST(NULL AS INT), CAST(NULL AS INT), 1);

SELECT * FROM T; -- output 1, 1, null, 1

如果希望读表时未被覆盖的字段有默认值而不是null，则需要fields.name.default-value。

CREATE TABLE T (
    k INT,
    a INT,
    b INT,
    c INT,
    PRIMARY KEY (k) NOT ENFORCED
) WITH (
    'merge-engine'='partial-update',
    'fields.b.default-value'='0'
);

INSERT INTO T VALUES (1, 1, CAST(NULL AS INT), CAST(NULL AS INT));
INSERT INTO T VALUES (1, CAST(NULL AS INT), CAST(NULL AS INT), 1);

SELECT * FROM T; -- output 1, 1, 0, 1

Aggregation

有时用户只关心聚合结果。 聚合 合并引擎根据聚合函数将同一主键下的各个值字段与最新数据一一聚合。 每个不属于主键的字段都可以被赋予一个聚合函数，由 fields.<field-name>.aggregate-function 表属性指定，否则它将使用 last_non_null_value 聚合作为默认值。 例如，请考虑下表定义。

CREATE TABLE MyTable (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT,
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.price.aggregate-function' = 'max',
    'fields.sales.aggregate-function' = 'sum'
);

price字段将通过 max 函数聚合，sales字段将通过 sum 函数聚合。 给定两个输入记录 <1, 23.0, 15> 和 <1, 30.2, 20>，最终结果将是 <1, 30.2, 35>。 当前支持的聚合函数和数据类型有：

• sum：支持 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT 和 DOUBLE。
• min/max：支持 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ。
• last_value / last_non_null_value：支持所有数据类型。
• listagg：支持STRING数据类型。
• bool_and / bool_or：支持BOOLEAN数据类型。
• first_value/first_not_null_value：支持所有数据类型。

只有 sum 支持撤回（UPDATE_BEFORE 和 DELETE），其他聚合函数不支持撤回。 如果允许某些函数忽略撤回消息，可以配置：'fields.${field_name}.ignore-retract'='true'

First Row

通过指定 'merge-engine' = 'first-row'，用户可以保留同一主键的第一行。 它与Deduplicate合并引擎不同，在First Row合并引擎中，它将生成仅insert changelog。

1. First Row合并引擎必须与 lookup changlog producer 一起使用。
2. 不能指定sequence.field。
3. 不接受 DELETE 和 UPDATE_BEFORE 消息。 可以配置 first-row.ignore-delete 来忽略这两种记录。

这对于替代流计算中的log deduplication有很大的帮助。

Changelog Producers

流式查询会不断产生最新的变化。 通过在创建表时指定更改changelog-producer表属性，用户可以选择从表文件生成的更改模式。

Changelog：通俗全面的理解就是操作过程中（比如ETL/CRUD），数据变化的日志；这样的日志可以帮助跟踪数据的历史变化，确保数据的质量与一致性，并允许回溯到之前的某个数据状态，帮助进行数据审计、数据分析、数据恢复等。

None

默认情况下，不会将额外的changelog producer应用于表的writer。 Paimon source只能看到跨snapshot的合并更改，例如删除了哪些键以及某些键的新值是什么。 但是，这些合并的更改无法形成完整的changelog，因为我们无法直接从中读取键的旧值。 合并的更改要求消费者“记住”每个键的值并重写这些值而不看到旧的值。 然而，一些消费者需要旧的值来确保正确性或效率。 考虑一个消费者计算某些分组键的总和（可能不等于主键）。 如果消费者只看到一个新值5，它无法确定应该将哪些值添加到求和结果中。 例如，如果旧值为 4，则应在结果中加 1。 但如果旧值是 6，则应依次从结果中减去 1。 旧的value对于这些类型的消费者来说很重要。 总而言之，没有一个changelog producer最适合数据库系统等使用者。 Flink 还有一个内置的"normalize"运算符，可以将每个键的值保留在状态中。 很容易看出，这种操作符的成本非常高，应该避免使用。 （可以通过“scan.remove-normalize”强制删除“normalize”运算符。）

Input

通过指定 'changelog- Producer' = 'input'，Paimon Writer依赖他们的输入作为完整changelog的来源。 所有输入记录将保存在单独的changelog file中，并由 Paimon source提供给消费者。 当 Paimon 编写者的输入是完整的changelog（例如来自数据库 CDC）或由 Flink 状态计算生成时，可以使用input changelog producer.

Lookup

如果您的输入无法生成完整的changelog，但想摆脱昂贵的"normalize"运算符，则可以考虑使用'lookup' changelog producer. 通过指定'changelog- Producer' = 'lookup'，Paimon将在提交数据写入之前通过'lookup'生成changelog。

Lookup 会将数据缓存在内存和本地磁盘上，您可以使用以下选项来调整性能：

Lookup changelog- Producer 支持changelog- Producer.row-deduplicate以避免为同一记录生成-U、+U changelog。

Full Compaction

如果你觉得“lookup”的资源消耗太大，可以考虑使用“full-compaction”changelog Producer，它可以解耦数据写入和changelog生成，更适合高延迟的场景（例如10分钟） ）。 通过指定 'changelog- Producer' = 'full-compaction'，Paimon 将比较完全压缩之间的结果并生成差异作为changelog。changelog的延迟受到完全压缩频率的影响。 通过指定 full-compaction.delta-commits 表属性，在增量提交（检查点 checkpoint）后将不断触发 full compaction。 默认情况下设置为 1，因此每个检查点都会进行完全压缩并生成change log。

Full-compaction changelog- Producer 支持changelog- Producer.row-deduplicate 以避免为同一记录生成-U、+U 变更日志。

Sequence Field

默认情况下，主键表根据输入顺序确定合并顺序（最后输入的记录将是最后合并的）。 然而在分布式计算中，会存在一些导致数据混乱的情况。 这时，可以使用时间字段作为sequence.field，例如：

CREATE TABLE MyTable (
    pk BIGINT PRIMARY KEY NOT ENFORCED,
    v1 DOUBLE,
    v2 BIGINT,
    dt TIMESTAMP
) WITH (
    'sequence.field' = 'dt'
);

无论输入顺序如何，具有最大sequence.field 值的记录将是最后合并的记录。 Sequence Auto Padding: 当记录更新或删除时，sequence.field必须变大，不能保持不变。 对于-U和+U，它们的序列字段必须不同。 如果您无法满足此要求，Paimon 提供了自动填充序列字段的选项。

1. 'sequence.auto-padding' = 'row-kind-flag'：如果对-U和+U使用相同的值，就像Mysql Binlog中的“op_ts”（数据库中进行更改的时间）一样。 建议使用自动填充行类型标志，它会自动区分-U（-D）和+U（+I）。
2. 精度不够：如果提供的sequence.field不满足精度，比如大约秒或毫秒，可以将sequence.auto-padding设置为秒到微或毫秒到微，这样序列号的精度 将由系统弥补到微秒。
3. 复合模式：例如“second-to-micro,row-kind-flag“，首先将micro添加到第二个，然后填充row-kind标志。

Row Kind Field

默认情况下，主键表根据输入行确定行类型。 您还可以定义“rowkind.field”以使用字段来提取行类型。 有效的行类型字符串应为“+I”、“-U”、“+U”或“-D”。