巨人肩膀
改版通知

巨人肩膀网站已全新改版。若您仍依赖旧站功能或数据，欢迎联系我们，我们会协助处理。联系我们

← 返回专栏列表

数据湖paimon入门指南

ckckck2025年1月10日142 浏览

主键表（Primary Key Table）

Merge Engines

sink-upsert-realize 可能会导致不正常的现象。当输入乱序时，建议使用序列字段来纠正无序。建议设置为 None。

sql 复制代码 
set table.exec.sink.upsert-materialize = NONE

Deduplicate

如果用户建表时不指定 merge-engine 配置，创建的 PK 表默认的 Merge Engine 是 deduplicate，即只保留最新的记录，其他的同 PK 数据则被丢弃。如果最新的记录是 DELETE 记录，那么相同 PK 的所有数据都将被删除。

sql 复制代码 
'merge-engine' = 'Deduplicate'

PartialUpdate

partial-update 必须跟 lookup 或者 full-compaction changelog producer 结合使用。Partial 无法接收 DELETE 消息，可以将 partial-update.ignore-delete 配置为忽略 delete 消息。

如果用户建表时指定 'merge-engine' = 'partial-update'，那么就会使用部分更新表引擎，可以做到多个 Flink 流任务去更新同一张表，每条流任务只更新一张表的部分列，最终实现一行完整的数据的更新。对于需要拉宽表的业务场景，partial-update 非常适合此场景，而且构建宽表的操作也相对简单。

sql 复制代码 
-- 创建 Partial Update 结果表
CREATE TABLE IF NOT EXISTS paimon.dw.order_detail (
    `order_id` string,
    `product_type` string,
    `plat_name` string,
    `ref_id` bigint,
    `start_city_name` string,
    `end_city_name` string,
    `create_time` timestamp(3),
    `update_time` timestamp(3),
    `dispatch_time` timestamp(3),
    `decision_time` timestamp(3),
    `finish_time` timestamp(3),
    `order_status` int,
    `binlog_time` bigint,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'bucket' = '20', -- 指定 20 个 bucket
    'bucket-key' = 'order_id',
    'sequence.field' = 'binlog_time', -- 记录排序字段
    'changelog-producer' = 'full-compaction', -- 选择 full-compaction，在 compaction 后产生完整的 changelog
    'changelog-producer.compaction-interval' = '2 min', -- compaction 间隔时间
    'merge-engine' = 'partial-update',
    'partial-update.ignore-delete' = 'true' -- 忽略 DELETE 数据，避免运行报错
);

Aggregation

如果用户建表时指定 'merge-engine' = 'aggregation'，此时使用聚合表引擎，可以通过聚合函数做一些预聚合。每个除主键以外的列都可以指定一个聚合函数，相同主键的数据就可以按照列字段指定的聚合函数进行相应的预聚合。如果不指定则默认为 last-non-null-value，空值不会覆盖。

Agg 表引擎也需要结合 lookup 或者 full-compaction 的 Changelog Producer 一起使用。需要注意的是，除了 SUM 函数，其他的 Agg 函数都不支持 Retraction。为了避免接收到 DELETE 和 UPDATE_BEFORE 消息报错，需要通过给指定字段配置 'fields.${field_name}.ignore-retract'='true' 忽略。

sql 复制代码 
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'
);

Change Producer

Changelog 主要应用在流读场景。流式查询将不断产生最新的更改。这些更改可以来自底层表文件，也可以来自像 Kafka 这样的外部日志系统。与外部日志系统相比，表文件的更改成本较低，但延迟较高（取决于创建快照的频率）。

通过在创建表时指定变更日志生产者表属性，用户可以选择从文件中生成的变更模式。

目前数仓分层是在 Paimon 里做的，数据以 Table Format 的形式存储在文件系统上。如果下游的 Flink 任务要流读 Paimon 表数据，需要存储帮助生成 Changelog（成本较低，但延迟相对较高），以便下游流读的，这时就需要我们在建表时指定 Paimon 的 Changelog Producer 决定以何种方式在何时生成 Changelog。如果不指定则不会在写入 Paimon 表的时候生成 Changelog，那么下游任务需要在流读时生成一个物化节点来产生 Changelog。这种方式的成本相对较高，同时官方不建议这样使用，因为下游任务在 State 中存储一份全量的数据，即每条数据以及其变更记录都需要保存在状态中。

Paimon 支持的 Changelog Producer 包括：

  • none：如果不指定，默认就是 none，成本较高，不建议使用。
  • input：如果我们的 Source 源是业务库的 Binlog，即写入 Paimon 表 Writer 任务的输入是完整的 Changelog，此时能够完全依赖输入端的 Changelog，并且将输入端的 Changelog 保存到 Paimon 的 Changelog 文件，由 Paimon Source 提供给下游流读。通过配置 'changelog-producer' = 'input'，将 Changelog Producer 设置为 input。
  • lookup：如果我们的输入不是完整的 Changelog，并且不想在下游流读时通过 Normalize 节点生成 Changelog，通过配置 'changelog-producer' = 'lookup'，通过 Lookup 的方式在数据写入的时候生成 Changelog。此 Changelog Producer 目前处于实验状态，暂未经过大量的生产验证。
  • full-compaction：除了以上几种方式，通过配置 'changelog-producer' = 'full-compaction' 将 Changelog Producer 设置为 full-compaction，Writer 端在 Compaction 后产生完整的 Changelog，并且写入到 Changelog 文件。通过设置 changelog-producer.compaction-interval 配置项控制 Compaction 的间隔和频率，不过此参数计划弃用，建议使用 full-compaction.delta-commits，此配置下默认为 1，即每次提交都做 Compaction。
sql 复制代码 
-- CREATE TABLE
CREATE TABLE t_dwd_table (
    id string,
    gn string,
    dt string,
    PRIMARY KEY (gn, id, log_create_unix_time, dt) NOT ENFORCED
) PARTITIONED BY (gn, dt) WITH (
    'bucket' = '8',
    'bucket-key' = 'id',
    'changelog-producer' = 'full-compaction',
    'changelog-producer.compaction-interval' = '54s',
    'snapshot.time-retained' = '24h'
);

仅追加表（Append Only Table）

Bucketing

未分区的表或分区表中的分区被细分为桶，为数据提供额外的结构，可用于更高效的查询。bucket 的范围由记录中一个或多个列的哈希值确定。用户可以通过提供 bucket-key 选项来指定 bucketing 列。如果未指定 bucket-key 选项，则主键（如果已定义）或完整记录将用作 bucket key。

存储桶是最小的读写存储单元，因此存储桶的数量限制了最大的处理并行性。不过，这个数字不应该太大，因为这会导致大量小文件和低读取性能。通常，每个 bucket 中的推荐数据大小约为 1GB。可以为仅追加表定义 bucket 编号。建议设置 bucket-key 字段。否则，数据将按照整行进行哈希，性能会很差。

sql 复制代码 
-- 建表时配置 'write-mode' = 'append-only'，用户可以创建 Append Only 表。
CREATE TABLE IF NOT EXISTS paimon.ods.event_log (
    ...
) PARTITIONED BY (...) WITH (
    'bucket' = '100',
    'bucket-key' = 'uuid',
    'snapshot.time-retained' = '7 d',
    'write-mode' = 'append-only'
);

INSERT INTO paimon.ods.event_log
SELECT ...
FROM realtime_event_kafka_source;

Compaction

Streaming Source

streaming source 目前仅在支持 Flink 引擎。

流读顺序

对于来自两个不同分区的任意两条记录：

  • 如果 scan.plan-sort-partition 设置为 true，则将首先生成分区值较小的记录。
  • 否则，将首先生成分区创建时间较早的记录。

对于来自同一分区和同一存储桶的任意两条记录，将首先生成第一条写入的记录。

对于来自同一分区但有两个不同存储桶的任意两条记录，不同的存储桶由不同的任务处理，它们之间没有顺序保证。

Watermark 定义

sql 复制代码 
CREATE TABLE T (
    `user` BIGINT,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (...);

-- 启动一个有界流作业来读取 paimon_table
SELECT window_start, window_end, COUNT(`user`) FROM TABLE(
    TUMBLE(TABLE T, DESCRIPTOR(order_time), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end;

有界流

Streaming Source 也可以是有界的，可以指定 scan.bounded.watermark 来定义有界流模式的结束条件，流读取将结束，直到遇到更大的水印快照。

sql 复制代码 
CREATE TABLE kafka_table (
    `user` BIGINT,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka'...);

-- 启动一个流式插入作业
INSERT INTO paimon_table SELECT * FROM kakfa_table;

-- 启动一个有界流作业来读取 paimon_table
SELECT * FROM paimon_table /*+ OPTIONS('scan.bounded.watermark'='...') */;

快照管理

快照过期

Paimon 写生成器每次提交都会生成一到两个快照。每个快照可以添加一些新的数据文件或将一些旧的数据文件标记为已删除。然而，标记的数据文件并没有被真正删除，因为 Paimon 还支持时间旅行到更早的快照。只有在快照过期时才会删除它们。

目前，Paimon 编写器在提交新更改时会自动执行过期操作。通过使旧快照过期，可以删除不再使用的旧数据文件和元数据文件以释放磁盘空间。

快照过期由以下表属性控制。

快照回滚

bash 复制代码 
<FLINK_HOME>/bin/flink run 
    /path/to/paimon-flink-action-0.5-SNAPSHOT.jar 
    rollback-to 
    --warehouse <warehouse-path> 
    --database <database-name> 
    --table <table-name> 
    --snapshot <snapshot-id>

文件布局

Snapshot Files

所有快照文件都存储在快照目录中。快照文件是一个 JSON 文件，包含有关此快照的信息，包括正在使用的 schema 文件、包含此快照的所有更改的清单列表。

Manifest Lists

所有清单列表和清单文件都存储在清单目录中。清单列表是清单文件名的列表。清单文件是包含有关 LSM 数据文件和变更日志文件的更改的文件。例如，在相应的快照中创建了哪个 LSM 数据文件，删除了哪个文件。

Data Files

数据文件按分区和存储桶分组。每个 bucket 目录都包含一个 LSM 树及其变更日志文件。目前，Paimon 支持使用 orc（默认）、parquet 和 avro 作为数据文件的格式。

LSM

Sorted-Runs

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

压缩

当越来越多的记录被写入 LSM 树时，排序的运行次数将增加。因为查询 LSM 树需要组合所有排序的运行，所以过多的排序运行将导致查询性能不佳，甚至内存不足。为了限制排序运行的数量，我们必须偶尔将几个排序运行合并为一个大的排序运行。此过程称为压实。

然而，压缩是一个资源密集型过程，需要消耗一定的 CPU 时间和磁盘 IO，因此过于频繁的压缩可能会导致写入速度变慢。这是查询性能和写入性能之间的权衡。Paimon 目前采用的压缩策略类似于 Rocksdb 的通用压缩。

默认情况下，当 Paimon 编写器将记录附加到 LSM 树时，他们还会根据需要执行压缩。用户还可以选择在专用压缩作业中执行所有压缩。有关更多信息，请参阅专用压实作业。

插入数据

sql 复制代码 
INSERT INTO T VALUES (1, 10001, 'varchar00001', '20230501');

删除数据

sql 复制代码 
DELETE FROM T WHERE dt >= '20230503';

表压缩（合并）

bash 复制代码 
<FLINK_HOME>/bin/flink run 
    /path/to/paimon-flink-action-0.5-SNAPSHOT.jar 
    compact 
    --warehouse <warehouse-path> 
    --database <database-name> 
    --table <table-name> 
    [--partition <partition-name>] 
    [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]]

过期快照

在 manifest 中被标记删除的记录不会立即被物理删除，而是要等到快照过期阶段判断可以安全删除才能和该快照一起被清理。

现在，我们以生产中常用的 CDC 导入为例，即通过 Flink CDC 流批一体读取 MySQL 全量和增量记录写入 Paimon，来串联以上提到的一系列文件操作。本节内容包括源端 CDC 数据的读取，Paimon 数据的写入和提交，异步小文件合并，和快照过期。

Java API

pom.xml

xml 复制代码 
<dependency>
    <groupId>org.apache.paimon</groupId>
    <artifactId>paimon-bundle</artifactId>
    <version>0.5-SNAPSHOT</version>
</dependency>

Create Catalog

java 复制代码 
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;

public class CreateCatalog {
    public static void createFilesystemCatalog() {
        CatalogContext context = CatalogContext.create(new Path("..."));
        Catalog catalog = CatalogFactory.createCatalog(context);
    }

    public static void createHiveCatalog() {
        // Paimon Hive catalog relies on Hive jars
        // You should add hive classpath or hive bundled jar.
        Options options = new Options();
        options.set("warehouse", "...");
        options.set("metastore", "hive");
        options.set("uri", "...");
        options.set("hive-conf-dir", "...");
        CatalogContext context = CatalogContext.create(options);
        Catalog catalog = CatalogFactory.createCatalog(context);
    }
}

RenameTable

java 复制代码 
import org.apache.paimon.fs.Path;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;

public class RenameTable {
    public static void main(String[] args) {
        Identifier fromTableIdentifier = Identifier.create("my_db", "my_table");
        Identifier toTableIdentifier = Identifier.create("my_db", "test_table");
        try {
            catalog.renameTable(fromTableIdentifier, toTableIdentifier, false);
        } catch (Catalog.TableAlreadyExistException e) {
            // do something
        } catch (Catalog.TableNotExistException e) {
            // do something
        }
    }
}

AlterTable

java 复制代码 
import org.apache.paimon.fs.Path;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import com.google.common.collect.Lists;
import java.util.Arrays;

public class AlterTable {
    public static void main(String[] args) {
        Identifier identifier = Identifier.create("my_db", "my_table");

        Map<String,String> options = new HashMap<>();
        options.put("bucket", "4");
        options.put("compaction.max.file-num", "40");

        catalog.createTable(
            identifier,
            new Schema(
                Lists.newArrayList(
                    new DataField(0, "col1", DataTypes.STRING(), "field1"),
                    new DataField(1, "col2", DataTypes.STRING(), "field2"),
                    new DataField(2, "col3", DataTypes.STRING(), "field3"),
                    new DataField(3, "col4", DataTypes.BIGINT(), "field4"),
                    new DataField(
                        4,
                        "col5",
                        DataTypes.ROW(
                            new DataField(5, "f1", DataTypes.STRING(), "f1"),
                            new DataField(6, "f2", DataTypes.STRING(), "f2"),
                            new DataField(7, "f3", DataTypes.STRING(), "f3")),
                        "field5"),
                    new DataField(8, "col6", DataTypes.STRING(), "field6")),
                Lists.newArrayList("col1"), // partition keys
                Lists.newArrayList("col1", "col2"), // primary key
                options,
                "table comment"),
            false);

        // add option
        SchemaChange addOption = SchemaChange.setOption("snapshot.time-retained", "2h");
        // remove option
        SchemaChange removeOption = SchemaChange.removeOption("compaction.max.file-num");
        // add column
        SchemaChange addColumn = SchemaChange.addColumn("col1_after", DataTypes.STRING());
        // add a column after col1
        SchemaChange.Move after = SchemaChange.Move.after("col1_after", "col1");
        SchemaChange addColumnAfterField = SchemaChange.addColumn("col7", DataTypes.STRING(), "", after);
        // rename column
        SchemaChange renameColumn = SchemaChange.renameColumn("col3", "col3_new_name");
        // drop column
        SchemaChange dropColumn = SchemaChange.dropColumn("col6");
        // update column comment
        SchemaChange updateColumnComment = SchemaChange.updateColumnComment(new String[]{"col4"}, "col4 field");
        // update nested column comment
        SchemaChange updateNestedColumnComment = SchemaChange.updateColumnComment(new String[]{"col5", "f1"}, "col5 f1 field");
        // update column type
        SchemaChange updateColumnType = SchemaChange.updateColumnType("col4", DataTypes.DOUBLE());
        // update column position, you need to pass in a parameter of type Move
        SchemaChange updateColumnPosition = SchemaChange.updateColumnPosition(SchemaChange.Move.first("col4"));
        // update column nullability
        SchemaChange updateColumnNullability = SchemaChange.updateColumnNullability(new String[]{"col4"}, false);
        // update nested column nullability
        SchemaChange updateNestedColumnNullability = SchemaChange.updateColumnNullability(new String[]{"col5", "f2"}, false);

        SchemaChange[] schemaChanges = new SchemaChange[] {
            addOption, removeOption, addColumn, addColumnAfterField, renameColumn, dropColumn,
            updateColumnComment, updateNestedColumnComment, updateColumnType, updateColumnPosition,
            updateColumnNullability, updateNestedColumnNullability};
        try {
            catalog.alterTable(identifier, Arrays.asList(schemaChanges), false);
        } catch (Catalog.TableNotExistException e) {
            // do something
        } catch (Catalog.TableAlreadyExistException e) {
            // do something
        } catch (Catalog.DatabaseNotExistException e) {
            // do something
        }
    }
}

Batch Read

java 复制代码 
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.types.DataTypes;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.util.List;

public class ReadTable {
    public static void main(String[] args) {
        // 1. Create a ReadBuilder and push filter (`withFilter`) and projection (`withProjection`) if necessary
        PredicateBuilder builder = new PredicateBuilder(RowType.of(DataTypes.STRING(), DataTypes.INT()));
        Predicate notNull = builder.isNotNull(0);
        Predicate greaterOrEqual = builder.greaterOrEqual(1, 12);

        ReadBuilder readBuilder = table.newReadBuilder()
                .withProjection(projection)
                .withFilter(Lists.newArrayList(notNull, greaterOrEqual));

        // 2. Plan splits in 'Coordinator' (or named 'Driver')
        List<Split> splits = readBuilder.newScan().plan().splits();

        // 3. Distribute these splits to different tasks

        // 4. Read a split in task
        TableRead read = readBuilder.newRead();
        RecordReader<InternalRow> reader = read.createReader(splits);
        reader.forEachRemaining(ReadTable::readRow);
    }
}

Batch Write

java 复制代码 
import java.util.List;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;

public class WriteTable {
    public static void main(String[] args) {
        // 1. Create a WriteBuilder (Serializable)
        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder()
                .withOverwrite(staticPartition);

        // 2. Write records in distributed tasks
        BatchTableWrite write = writeBuilder.newWrite();

        GenericRow record1 = GenericRow.of(BinaryString.fromString("Alice"), 12);
        GenericRow record2 = GenericRow.of(BinaryString.fromString("Bob"), 5);
        GenericRow record3 = GenericRow.of(BinaryString.fromString("Emily"), 18);

        write.write(record1);
        write.write(record2);
        write.write(record3);

        List<CommitMessage> messages = write.prepareCommit();

        // 3. Collect all CommitMessages to a global node and commit
        BatchTableCommit commit = writeBuilder.newCommit();
        commit.commit(messages);

        // Abort unsuccessful commit to delete data files
        // commit.abort(messages);
    }
}

Stream Read

java 复制代码 
import java.io.IOException;
import java.util.List;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.table.source.TableRead;

public class StreamReadTable {
    public static void main(String[] args) throws IOException {
        // 1. Create a ReadBuilder and push filter (`withFilter`) and projection (`withProjection`) if necessary
        ReadBuilder readBuilder = table.newReadBuilder()
                .withProjection(projection)
                .withFilter(filter);

        // 2. Plan splits in 'Coordinator' (or named 'Driver')
        StreamTableScan scan = readBuilder.newStreamScan();
        while (true) {
            List<Split> splits = scan.plan().splits();
            // Distribute these splits to different tasks

            Long state = scan.checkpoint();
            // can be restored in scan.restore(state) after failover
        }

        // 3. Read a split in task
        TableRead read = readBuilder.newRead();
        RecordReader<InternalRow> reader = read.createReader(splits);
        reader.forEachRemaining(row -> System.out.println(row));
    }
}

Stream Write

exactly-once 关键保证：

  • CommitUser 表示一个用户。一个用户可以提交多次。在分布式处理中，您应该使用相同的 commitUser。
  • 不同的应用程序需要使用不同的 commitUsers。
  • StreamTableWrite 和 StreamTableCommit 的 commitIdentifier 需要一致，并且需要为下一次提交增加 id。
  • 发生故障时，如果仍有未提交的 CommitMessages，请使用 StreamTableCommit#filterCommitted 通过 commitIdentifier 排除已提交的消息。
java 复制代码 
import java.util.List;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;

public class StreamWriteTable {
    public static void main(String[] args) throws Exception {
        // 1. Create a WriteBuilder (Serializable)
        StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();

        // 2. Write records in distributed tasks
        StreamTableWrite write = writeBuilder.newWrite();
        // commitIdentifier like Flink checkpointId
        long commitIdentifier = 0;
        while (true) {
            write.write(record1);
            write.write(record2);
            write.write(record3);
            List<CommitMessage> messages = write.prepareCommit(false, commitIdentifier);
            commitIdentifier++;
        }

        // 3. Collect all CommitMessages to a global node and commit
        StreamTableCommit commit = writeBuilder.newCommit();
        commit.commit(commitIdentifier, messages);

        // 4. When failover, you can use 'filterCommitted' to filter committed commits.
        commit.filterCommitted(committedIdentifiers);
    }
}

Flink API

pom.xml

xml 复制代码 
<dependency>
    <groupId>org.apache.paimon</groupId>
    <artifactId>paimon-flink-1.17</artifactId>
    <version>0.5-SNAPSHOT</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_2.12</artifactId>
    <version>1.17.0</version>
    <scope>provided</scope>
</dependency>

Read from Table

java 复制代码 
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class ReadFromTable {
    public static void readFrom() throws Exception {
        // create environments of both APIs
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create paimon catalog
        tableEnv.executeSql("CREATE CATALOG paimon WITH ('type' = 'paimon', 'warehouse'='...')");
        tableEnv.executeSql("USE CATALOG paimon");

        // convert to DataStream
        Table table = tableEnv.sqlQuery("SELECT * FROM my_paimon_table");
        DataStream<Row> dataStream = tableEnv.toChangelogStream(table);

        // use this datastream
        dataStream.executeAndCollect().forEachRemaining(System.out::println);

        // prints:
        // +I[Bob, 12]
        // +I[Alice, 12]
        // -U[Alice, 12]
        // +U[Alice, 14]
    }
}

Write to Table

java 复制代码 
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

public class WriteToTable {
    public static void writeTo() {
        // create environments of both APIs
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create a changelog DataStream
        DataStream<Row> dataStream = env.fromElements(
            Row.ofKind(RowKind.INSERT, "Alice", 12),
            Row.ofKind(RowKind.INSERT, "Bob", 5),
            Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
            Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100)
        ).returns(Types.ROW_NAMED(new String[] {"name", "age"}, Types.STRING, Types.INT));

        // interpret the DataStream as a Table
        Schema schema = Schema.newBuilder()
            .column("name", DataTypes.STRING())
            .column("age", DataTypes.INT())
            .build();
        Table table = tableEnv.fromChangelogStream(dataStream, schema);

        // create paimon catalog
        tableEnv.executeSql("CREATE CATALOG paimon WITH ('type' = 'paimon', 'warehouse'='...')");
        tableEnv.executeSql("USE CATALOG paimon");

        // register the table under a name and perform an aggregation
        tableEnv.createTemporaryView("InputTable", table);

        // insert into paimon table from your data stream table
        tableEnv.executeSql("INSERT INTO sink_paimon_table SELECT * FROM InputTable");
    }
}

Read Performance

完全压缩

配置 full-compaction.delta-commits 在 Flink 写入中定期执行完全压缩。并且它可以确保分区在写入结束之前被完全压缩。建议不要设置超过快照过期时间（默认为 1 小时）的值。例如，如果检查点间隔为 1 分钟，则建议将 full-compaction.delta-commits 设置为 30。

主键表

对于主键表，它是一种 MOR（MergeOnRead）技术。在读取数据时，多层 LSM 数据被合并，并行性的数量将受到桶的数量的限制。尽管 Paimon 的合并会很有效，但它仍然无法赶上普通的 AppendOnly 表。

如果希望在某些情况下查询足够快，但只能找到较旧的数据，您可以：

  • 配置 full-compaction.delta-commits，在写入数据时（当前仅为 Flink），将定期执行完全压缩。
  • 将 scan.mode 配置为 compacted-full，读取数据时，将拾取完全压缩的快照。读取性能良好。

可以在读取时灵活地平衡查询性能和数据延迟。

仅追加表

小文件可能会减慢读取速度并影响 DFS 的稳定性。默认情况下，当单个存储桶中的小文件超过 compaction.max.file-num（默认为 50 个）时，会触发压缩。但是，当存在多个 bucket 时，会生成许多小文件。可以使用完全压缩来减少小文件。完全压缩将消除大多数小文件。

Write Performance

并行度

写入初始化

在写入初始化过程中，bucket 的写入程序需要读取所有历史文件。如果这里存在瓶颈（例如，同时写入大量分区），可以使用 write-manifest-cache 来缓存读取的清单数据，以加速初始化。

压缩

  • 触发压缩的排序运行数
  • 要暂停写入的排序运行数
  • 全量合并作业

Dedicated Compaction Job

可能会导致写吞吐量不稳定，因为在执行压缩时吞吐量可能会暂时下降。压缩会将一些数据文件标记为“已删除”（并非真正删除，有关更多信息，请参阅过期快照）。如果多个写入程序标记同一个文件，则在提交更改时会发生冲突。Paimon 会自动解决冲突，但这可能会导致作业重新启动。

bash 复制代码 
<FLINK_HOME>/bin/flink run 
    /path/to/paimon-flink-action-0.5-SNAPSHOT.jar 
    compact 
    --warehouse <warehouse-path> 
    --database <database-name> 
    --table <table-name> 
    [--partition <partition-name>] 
    [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]]

内存

  • 写入程序的内存缓冲区（memory buffer），由单个任务的所有写入程序共享并抢占。此内存值可以通过 write-buffer-size 进行调整。
  • 合并几个排序运行以进行压缩时消耗的内存。可以通过 num-sorted-run.compaction-tigger 选项进行调整，以更改要合并的排序运行数。
  • 如果行非常大，在进行压缩时，一次读取太多行数据可能会消耗大量内存。减少 read.batch-size 选项可以减轻这种情况的影响。
  • 写入不可调整的柱状（ORC、Parquet 等）文件所消耗的内存。

数据写入 Paimon 的流程

end