Paimon的下载及安装，并且了解了主键表的引擎以及changelog-producer的含义参考：

• 大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(1)

利用Paimon表做lookup join，集成mysql cdc等参考：

• 大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(2)

利用Paimon的Tag兼容Hive，Branch管理等参考:

• 大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(3)

大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(3)

今天，我们继续快速了解下最近比较火的Apache Paimon：

• 官方文档：https://paimon.apache.org/docs/1.0/
• 推荐阅读：当流计算邂逅数据湖：Paimon 的前生今世

1 利用Tag兼容Hive

• Paimon 的每一次写都会生成一个 Immutable 的快照，快照可以被 Time Travel 的读取。
• 但在大多数情况下，作业会生成过多的快照，所以根据表配置，快照会在合适的时间点被过期。快照过期还会删除旧的数据文件，过期快照的历史数据将无法再查询。
• 要解决此问题，可以基于快照创建 Tag。Tag 将维护快照的清单和数据文件。
  • 典型的用法是每天创建Tag(如下图所示)，然后可以维护每天的历史数据以进行批式查询。
  • 推荐在 ODS 层使用 Tag 来替代 Hive 的分区，但是后续的 DWD 和 DWS 不建议。

1.1 Tag创建

1.1.1 自动创建

-- Flink SQL
CREATE TABLE t (
    k INT PRIMARY KEY NOT ENFORCED,
    f0 INT,
    ...
) WITH (
    'tag.automatic-creation' = 'process-time',  -- 基于process-time自动创建
    'tag.creation-period' = 'daily',            -- 创建间隔：每天
    'tag.creation-delay' = '10 m',              -- 延迟10min
    'tag.num-retained-max' = '90'               -- 最大保存90天
);

• 上面配置表明每天0点10分钟创建一个 Tag，最大保留3个月的 Tag，Flink 流式写入，自动创建 Tags，自动清理 Tags。

1.1.2 利用Action包创建Tag

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.0.0.jar \
    create_tag \
    --warehouse <warehouse-path> \
    --database <database-name> \ 
    --table <table-name> \
    --tag_name <tag-name> \
    [--snapshot <snapshot_id>] \
    [--time_retained <time-retained>] \
    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]

• 如果未设置snapshot, 那么默认snapshot_id默认为最新
• 当然，也能删除Tag、回滚Tag，可以参考官网命令：
  • Manage Tags | Apache Paimon

1.2 利用Tag映射

• 有了 Tag 后，可以在 Flink SQL 或者 Spark SQL 里使用 Time Travel 来查询 Tags；

• 但是这给业务带来了一个问题，老的 Hive SQL 如何兼容？老的 Hive 可是一个全量分区表，而 Paimon 表是一个非分区主键表，Hive 数据仓库的传统使用更习惯于使用分区来指定查询的 Tag。

• paimon引入了 metastore.tag-to-partition 和 metastore.tag-to-partition.preview'(配置此参数可以让 Hive SQL 查询到未 Tag 的分区，比如当前最新数据) 来将未分区的主键表映射到 Hive metastore 中的分区表，并映射分区字段为 Tag 查询。

• Flink 结合 Paimon 打造的入湖架构如下：

  • 通过 Flink CDC 一键全增量一体入湖到 Paimon，此任务可以配置 Tag 的自动创建，然后通过 Paimon 的能力，将 Tag 映射为 Hive 的分区，完全兼容原有 Hive SQL 的用法。
  • 优势如下：
    • 架构链路复杂度低，不再因为各种组件的问题导致链路延时，你只用运维这一个流作业，而且可以完全兼容原有 Hive SQL 用法。
    • 时延低：延时取决于流作业的 Checkpoint Interval，数据最低1分钟实时可见 (建议1-5分钟)。不但如此，Paimon 也提供了流读的能力，让你完成分钟级的 Streaming 计算，也可以写到下游别的存储。
    • 存储成本低：得益于湖格式的 Snapshot 管理，加上 LSM 的文件复用，比如同样是存储 100天的快照，原有 Hive 数仓 100 天需要 100 份的存储，Paimon 在某些增量数据不多的场景只需要 2 份的存储，大幅节省存储资源。
    • 计算成本低：得益于 LSM 的增量合并能力，此条链路只有增量数据的处理，没有全量的合并。可能有用户会担心，常驻的流作业会消耗更多的资源，对 Paimon 来说，你可以打开纯异步 Compaction 的机制，以 Paimon 优异的性能表现，只用少量的资源即可完成同步，Paimon 另有整库同步等能力帮助节省资源。

1.2.1 tag-to-partition

-- 创建映射的paimon表
Flink SQL> drop table if exists mydb_t;
Flink SQL> CREATE TABLE mydb_t (
    pk INT,
    col1 STRING,
    col2 STRING
) WITH (
    'bucket' = '-1',
    -- Only Hive Engine can be used to query these upsert-to-partitioned tables.
    -- 将tag映射为hive分区
    'metastore.tag-to-partition' = 'dt' 
);

-- 插入数据
-- snapshot=1
Flink SQL> INSERT INTO mydb_t VALUES (1, '10', '100'), (2, '20', '200');
-- snapshot=2
Flink SQL> INSERT INTO mydb_t VALUES (3, '30', '300'), (4, '40', '400');

• 然后，利用action包创建Tag

# 利用action 包创建tag\
# 依旧利用hive元数据做catalog
[root@centos01 ~]# /opt/apps/flink-1.16.0/bin/flink run \
    /opt/apps/flink-1.16.0/lib/paimon-flink-action-1.0.0.jar \
    create_tag \
    --warehouse hdfs://centos01:8020/user/hive/warehouse \
    --database paimon_db \
    --table mydb_t \
    --tag_name '2025-02-18' \
	--catalog-conf metastore=hive \
    --catalog-conf uri=thrift://centos01:9083 \
    --snapshot 1
    
[root@centos01 ~]# /opt/apps/flink-1.16.0/bin/flink run \
    /opt/apps/flink-1.16.0/lib/paimon-flink-action-1.0.0.jar \
    create_tag \
    --warehouse hdfs://centos01:8020/user/hive/warehouse \
    --database paimon_db \
    --table mydb_t \
    --tag_name '2025-02-19' \
	--catalog-conf metastore=hive \
    --catalog-conf uri=thrift://centos01:9083 \
    --snapshot 2 

• 我们就可以在hive中查看分区，且查询数据

0: jdbc:hive2://192.168.42.101:10000> show partitions mydb_t;
+----------------+
|   partition    |
+----------------+
| dt=2025-02-18  |
| dt=2025-02-19  |
+----------------+
2 rows selected (0.438 seconds)

0: jdbc:hive2://192.168.42.101:10000> select * from mydb_t a where dt = '2025-02-18';
+-------+---------+---------+-------------+
| a.pk  | a.col1  | a.col2  |    a.dt     |
+-------+---------+---------+-------------+
| 1     | 10      | 100     | 2025-02-18  |
| 2     | 20      | 200     | 2025-02-18  |
+-------+---------+---------+-------------+
2 rows selected (3.27 seconds)

0: jdbc:hive2://192.168.42.101:10000> select * from mydb_t a where dt = '2025-02-19';
+-------+---------+---------+-------------+
| a.pk  | a.col1  | a.col2  |    a.dt     |
+-------+---------+---------+-------------+
| 1     | 10      | 100     | 2025-02-19  |
| 2     | 20      | 200     | 2025-02-19  |
| 3     | 30      | 300     | 2025-02-19  |
| 4     | 40      | 400     | 2025-02-19  |
+-------+---------+---------+-------------+

1.2.2 tag-to-partition.preview

• 上述示例只能查询已经创建的tag，但Paimon是一个实时数据湖，您还需要查询最新的数据。因此，Paimon提供了一个预览功能
• 'metastore.tag-to-partition.preview'可选值如下：
  • “none”：不自动创建标签；
  • “process-time”：基于机器时间，当处理时间超过周期时间加上延迟时，创建标签；
  • “watermark”：基于输入的watermark，当watermark超过周期时间加上延迟时，创建标签；
  • “batch”：在批处理场景中，任务完成后生成当前快照对应的标签。

Flink SQL> drop table if exists mydb_preview;
Flink SQL> CREATE TABLE mydb_preview (
    pk INT,
    col1 STRING,
    col2 STRING
) WITH (
    'bucket' = '-1',
    'metastore.tag-to-partition' = 'dt',
    -- paimon会基于process-time提前创建partitions
    'metastore.tag-to-partition.preview' = 'process-time'
);

-- snapshot=1
Flink SQL> INSERT INTO mydb_preview VALUES (1, '10', '100'), (2, '20', '200');

-- create tag '2025-02-19' for snapshot 1
[root@centos01 ~]# /opt/apps/flink-1.16.0/bin/flink run \
    /opt/apps/flink-1.16.0/lib/paimon-flink-action-1.0.0.jar \
    create_tag \
    --warehouse hdfs://centos01:8020/user/hive/warehouse \
    --database paimon_db \
    --table mydb_preview \
    --tag_name '2025-02-19' \
	--catalog-conf metastore=hive \
    --catalog-conf uri=thrift://centos01:9083 \
    --snapshot 1

0: jdbc:hive2://192.168.42.101:10000> show partitions mydb_preview;;
+----------------+
|   partition    |
+----------------+
| dt=2025-02-19  |
| dt=2025-02-20  |
+----------------+
2 rows selected (0.085 seconds)
0: jdbc:hive2://192.168.42.101:10000> select * from mydb_preview a where dt = '2025-02-19';
+-------+---------+---------+-------------+
| a.pk  | a.col1  | a.col2  |    a.dt     |
+-------+---------+---------+-------------+
| 1     | 10      | 100     | 2025-02-19  |
| 2     | 20      | 200     | 2025-02-19  |
+-------+---------+---------+-------------+
2 rows selected (0.292 seconds)
0: jdbc:hive2://192.168.42.101:10000> select * from mydb_preview a where dt = '2025-02-20';
+-------+---------+---------+-------------+
| a.pk  | a.col1  | a.col2  |    a.dt     |
+-------+---------+---------+-------------+
| 1     | 10      | 100     | 2025-02-20  |
| 2     | 20      | 200     | 2025-02-20  |
+-------+---------+---------+-------------+
2 rows selected (0.263 seconds)



-- new data in '2025-02-20'
Flink SQL> INSERT INTO mydb_preview VALUES (3, '30', '300'), (4, '40', '400');

0: jdbc:hive2://192.168.42.101:10000> select * from mydb_preview a where dt = '2025-02-20';
+-------+---------+---------+-------------+
| a.pk  | a.col1  | a.col2  |    a.dt     |
+-------+---------+---------+-------------+
| 1     | 10      | 100     | 2025-02-20  |
| 2     | 20      | 200     | 2025-02-20  |
| 3     | 30      | 300     | 2025-02-20  |
| 4     | 40      | 400     | 2025-02-20  |
+-------+---------+---------+-------------+

2 Branch分支管理

• 在流式数据处理中，修正数据具有挑战性，因为它可能会影响现有数据，而用户会看到流式的临时结果，这是不期望的。
• 我们假设现有工作流正在处理的分支是main分支。通过创建自定义数据分支，可以在现有表上对新任务进行实验性测试和数据验证，而无需停止现有的读取/写入工作流，也无需从主分支复制数据。
• 通过合并或替换分支操作，用户可以完成数据的修正。

-- 1、创建paimon表
Flink SQL> drop table if exists flink_branch_demo;
Flink SQL> CREATE TABLE flink_branch_demo (
    dt STRING NOT NULL,
    name STRING NOT NULL,
    amount BIGINT,
    PRIMARY KEY (dt, name) NOT ENFORCED 
) PARTITIONED BY (dt)
WITH (
    'connector' = 'paimon'
);

-- 2、创建一个专门用于流写的分支streambranch, 这个分支将负责接收实时流入的数据。
[root@centos01 ~]# /opt/apps/flink-1.16.0/bin/flink run \
    /opt/apps/flink-1.16.0/lib/paimon-flink-action-1.0.0.jar \
    create_branch \
    --warehouse hdfs://centos01:8020/user/hive/warehouse \
    --database paimon_db \
    --table flink_branch_demo \
    --branch_name streambranch \
    --catalog-conf metastore=hive \
    --catalog-conf uri=thrift://centos01:9083


-- 3、设置流写分支的属性
Flink SQL> ALTER TABLE  `flink_branch_demo$branch_streambranch` SET (
    'bucket' = '4',
    'changelog-producer' = 'lookup'
);

-- 4、设置回滚分支（如果要实现分支回滚必须要设置该参数)
Flink SQL> ALTER TABLE flink_branch_demo SET ( 'scan.fallback-branch' = 'streambranch' );

-- 5、写入数据
-- 5-1、主分支写入数据
Flink SQL> insert into flink_branch_demo values  ('20240725', 'apple', 3), ('20240725', 'banana', 5);

Flink SQL> select * from flink_branch_demo;
+----------+--------+--------+
|       dt |   name | amount |
+----------+--------+--------+
| 20240725 |  apple |      3 |
| 20240725 | banana |      5 |
+----------+--------+--------+
2 rows in set

--  5-2、再往streambranch分支写入数据
Flink SQL> INSERT INTO `flink_branch_demo$branch_streambranch` 
VALUES ('20240725', 'apple', 666), ('20240725', 'peach', 999), ('20240726', 'cherry', 33), ('20240726', 'pear', 88);


-- 5-3、查询主分支
-- 20240725分区的新的数据没有生效！ 那说明原表已经有的分区的数据，在streambranch写入这些分区的数据，原表是不会更新的，只要是往原表里面写了某个分区的数据，那么这个分区的数据以写入原表主分支的为准。
-- 原表主分支没有的分区的数据，则按照streambranch读取，因为设置了原表的 'scan.fallback-branch' = 'streambranch' ，读取原表可以查到streambranch这部分的数据。
Flink SQL> select * from flink_branch_demo;
+----------+--------+--------+
|       dt |   name | amount |
+----------+--------+--------+
| 20240726 | cherry |     33 | -- 26号分区主表没有，使用了分支表中的数据
| 20240726 |   pear |     88 |
| 20240725 |  apple |      3 | -- 25号的分区使用了主表中的数据
| 20240725 | banana |      5 |
+----------+--------+--------+

-- 5-4、查询流分支
Flink SQL> select * from `flink_branch_demo$branch_streambranch` ;
+----------+--------+--------+
|       dt |   name | amount |
+----------+--------+--------+
| 20240726 | cherry |     33 |
| 20240726 |   pear |     88 |
| 20240725 |  apple |    666 |
| 20240725 |  peach |    999 |
+----------+--------+--------+
4 rows in set


-- 6、合并分支
-- 合并分支表操作(Fast Forward)，即：删除主表的一切数据，并将分支表的一切数据拷贝到主表
[root@centos01 ~]# /opt/apps/flink-1.16.0/bin/flink run \
    /opt/apps/flink-1.16.0/lib/paimon-flink-action-1.0.0.jar \
    fast_forward \
    --warehouse hdfs://centos01:8020/user/hive/warehouse \
    --database paimon_db \
    --table flink_branch_demo \
    --branch_name streambranch \
    --catalog-conf metastore=hive \
    --catalog-conf uri=thrift://centos01:9083

-- 再次查询主表
Flink SQL> select * from flink_branch_demo;
+----------+--------+--------+
|       dt |   name | amount |
+----------+--------+--------+
| 20240726 | cherry |     33 |
| 20240726 |   pear |     88 |
| 20240725 |  apple |    666 |
| 20240725 |  peach |    999 |
+----------+--------+--------+


-- 7、数据回归到主分支版本(注意：不进行上面合并操作)
Flink SQL> ALTER TABLE flink_branch_demo RESET( 'scan.fallback-branch');

3 追加表(Append table)

• 如果一个表没有定义主键，那它就是一个追加表（Append Table）。与主键表相比，追加表无法直接接收变更日志，也不能直接通过 upsert 更新数据，只能接收追加数据。

使用场景或优势	说明
批量写入和批量读取	类似于常规的 Hive 分区表，适用于大规模数据的批量处理。
友好的对象存储	良好的兼容性和适应性，支持 S3、OSS 等对象存储。
时间穿越和回滚	支持数据的时间旅行和回滚功能，方便数据的历史查询和恢复。
低成本的删除和更新	在批量数据操作中，能够以较低的计算和资源成本进行删除和更新操作。
流式接收中的小文件自动合并	在流式写入过程中，自动处理小文件合并，减少存储碎片。
队列形式的流式读写	支持如队列般的流式读写操作，可以像消息队列一样处理数据。
高性能查询	通过顺序和索引实现的高效查询性能。

3.1 流式处理

• Append Table可以通过 Flink 进行非常灵活的流式写入，并可以像队列一样通过 Flink 进行读取。
• 唯一的区别是其延迟为分钟级别，但其优势在于非常低的成本以及能够进行过滤和投影下推。

3.1.1 小文件自动合并

• 在流式写入作业中，如果没有定义分桶（bucket），写入器不会进行压缩；
• 相反，将使用压缩协调器（Compact Coordinator）扫描小文件并将压缩任务传递给压缩工作者（Compact Worker）。
• 流式模式下，如果在 Flink 中运行插入 SQL，拓扑结构将如下所示：

• 注意：
  • 上面的压缩任务不会引起反压。
  • 如果设置 write-only 为 true，压缩协调器（Compact Coordinator）和压缩工作者（Compact Worker）将在拓扑中被移除。
  • 自动压缩仅在 Flink 引擎的流模式下被支持。可以通过 Paimon 在 Flink 中启动压缩作业，并通过设置 write-only 禁用所有其他压缩。

3.1.2 流式查询

• 追加表可以像消息队列一样使用，进行流式查询，与主键表类似，有两个选项可以进行流式读取：
  • 默认模式：流式读取在首次启动时生成表的最新快照，并继续读取最新的增量记录。
  • 增量模式：可以指定 scan.mode 或 scan.snapshot-id 或 scan.timestamp-millis 或 scan.file-creation-time-millis 进行增量读取。
• 追加表的流式查询类似 Flink-Kafka，默认情况下不保证顺序。如果数据需要某种顺序，也需要考虑定义桶键（bucket-key），即Bucketed Append。

3.2 查询优化

3.2.1 按照顺序跳过查询

• Paimon 默认在清单文件中记录每个字段的最大值和最小值。
• 在查询时，根据查询的 WHERE 条件，通过清单中的统计信息进行文件过滤。如果过滤效果良好，查询时间可以从分钟级别加速到毫秒级别。
• 然而，数据分布并不总是能有效过滤，因此如果可以根据 WHERE 条件中的字段对数据进行排序，将会更高效。
  • 具体可参考：Flink COMPACT Action or Flink COMPACT Procedure or Spark COMPACT Procedure.

3.2.2 按文件索引跳过数据

• 如下代码所示，可以使用文件索引，会在读取端通过索引过滤文件

  • 定义 file-index.bloom-filter.columns 后，Paimon 将为每个文件创建相应的索引文件。
  • 如果索引文件太小，它将直接存储在清单中，否则将存储在数据文件的目录中。
  • 每个数据文件对应一个索引文件，该文件有独立的定义，可以包含不同类型的多列索引。

  CREATE TABLE my_table (
      product_id BIGINT,
      price DOUBLE,
      sales BIGINT
  ) WITH (
      'file-index.bloom-filter.columns' = 'product_id',
      'file-index.bloom-filter.product_id.items' = '200'
  );
  

• 索引种类如下所示：

  # 布隆过滤器索引
  file-index.bloom-filter.columns：指定需要创建布隆过滤器索引的列。
  file-index.bloom-filter.<column_name>.fpp：配置布隆过滤器的误报率（False Positive Probability）。
  file-index.bloom-filter.<column_name>.items：配置每个数据文件中预期的唯一项数量。
  
  # Bitmap（位图索引）：
  file-index.bitmap.columns：指定需要创建位图索引的列。
  
  
  # Bit-Slice Index Bitmap（位切片索引位图）：
  file-index.bsi.columns：指定需要创建位切片索引（BSI）的列。
  
  
  如果想为现有表添加文件索引，且不进行任何数据重写，可以使用rewrite_file_index过程。
  在使用该过程之前，可以使用ALTER子句来为表配置file-index.<filter-type>.columns。
  可以参考：
  https://paimon.apache.org/docs/1.0/flink/procedures/#procedures
  

  • 布隆过滤器索引和位图索引的区别

  特性	布隆过滤器索引（Bloom Filter Index）	位图索引（Bitmap Index）
  设计目标	快速判断某个值是否可能存在，减少磁盘 I/O	精确查询低基数列，支持多条件组合查询
  实现原理	基于哈希函数的概率型数据结构	基于Bitmap的精确索引结构
  适用数据类型	高基数列（如唯一 ID、字符串等）	低基数列（如性别、状态等）
  查询类型	等值查询（=）	等值查询（=）和多条件组合查询（AND、OR）
  存储效率	存储空间小，适合大规模数据集	低基数列存储效率高，高基数列存储开销大
  查询效率	查询速度快，但存在误报率	查询效率高，无误报率
  更新代价	较低	较高
  适用场景	大数据集的快速过滤	低基数列的精确查询和多条件组合查询

  • 位图索引和位切片索引的区别

  特性	Bitmap Index（位图索引）	Bit-Slice Index (BSI)（位切片索引）
  适用数据类型	低基数(即列中唯一值的数量较少)的任意类型（如枚举、状态等）	高基数的数值型数据（如金额、时间戳等）
  查询类型	等值查询、范围查询	范围查询、聚合查询（如 SUM、MAX 等）
  存储效率	低基数列高效，高基数列存储开销大	高基数列存储效率高
  实现复杂度	简单	复杂
  更新代价	较高	较高

3.3 Bucketed Append

• 可以指定 bucket 和 bucket-key 以创建一个Bucketed Append表。
• 在Bucketed Append中，不同桶内的数据是严格有序的，流式读取将按写入顺序准确地传输记录。这样可以优化数据处理和查询性能。

-- 创建Bucketed Append表
CREATE TABLE my_table (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT
) WITH (
    'bucket' = '8',
    'bucket-key' = 'product_id'
);

3.3.1 有界流

• 流式来源（Streaming Source）也可以是有界的，可以通过指定 scan.bounded.watermark 来定义有界流模式的结束条件。
• 例如，指定kafka源并声明watermark 的定义。当使用此kafka源写入Paimon表时，Paimon表的快照将生成相应的watermark，以便流式读取此Paimon表时可以使用有界watermark的功能。

-- 临时表
drop TEMPORARY table if exists order_from_kafka;
CREATE TEMPORARY TABLE order_from_kafka (
    `user` int,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '8' HOUR - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders_test',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
	'properties.group.id' = 'tGroup',
	'json.fail-on-missing-field' = 'false',
	'scan.startup.mode' = 'earliest-offset',
	'json.ignore-parse-errors' = 'true'
);

-- 创建topic
/opt/apps/kafka_2.12-2.6.2/bin/kafka-topics.sh --create --topic orders_test --replication-factor 1 --partitions 1 --bootstrap-server centos01:9092 

-- paimon追加表
drop table if exists paimon_r;
CREATE TABLE paimon_r (
    `user` int,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '8' HOUR - INTERVAL '5' SECOND
) WITH (
    'connector' = 'paimon'
);


-- 将Kafka表中的数据实时插入到Paimon表中：
INSERT INTO paimon_r SELECT * FROM order_from_kafka;

-- 启动有界流任务读取 Paimon 表
-- 1696126500000 2023-10-01 10:15:00

-- 当Flink处理过程中遇到第一个水印值大于或等于这个时间点的记录时，
-- 它会停止继续读取后续的数据，即使数据源中还有更晚时间点的数据。
Flink SQL> SELECT * FROM paimon_r /*+ OPTIONS('scan.bounded.watermark'='1696126500000') */;
+----+-------------+--------------------------------+-------------------------+
| op |        user |                        product |              order_time |
+----+-------------+--------------------------------+-------------------------+
| +I |        1001 |                      iPhone 15 | 2023-10-01 10:00:00.000 |
| +I |        1002 |                    MacBook Pro | 2023-10-01 10:05:00.000 |
| +I |        1003 |                    AirPods Pro | 2023-10-01 10:10:00.000 |
| +I |        1004 |                       iPad Air | 2023-10-01 10:15:00.000 |
+----+-------------+--------------------------------+-------------------------+
Received a total of 4 rows


-- 启动命令行生产者，模拟数据源源源不断地生产数据(每隔一段时间插入1条数据)
/opt/apps/kafka_2.12-2.6.2/bin/kafka-console-producer.sh --topic orders_test --bootstrap-server centos01:9092
{"user": 1001, "product": "iPhone 15", "order_time": "2023-10-01 10:00:00"}
{"user": 1002, "product": "MacBook Pro", "order_time": "2023-10-01 10:05:00"}
{"user": 1003, "product": "AirPods Pro", "order_time": "2023-10-01 10:10:00"}
-- "2023-10-01 10:15:00" 时候watermark是1696126495000，即：2023-10-01 10:14:55
-- 此时有界流并未结束
{"user": 1004, "product": "iPad Air", "order_time": "2023-10-01 10:15:00"}
-- "2023-10-01 10:20:00" 时候watermark是1696126795000
-- 即：2023-10-01 10:19:55 > 2023-10-01 10:15:00(1696126500000)
-- 停止继续读取后续的数据，即使数据源中还有更晚时间点的数据
{"user": 1005, "product": "Apple Watch", "order_time": "2023-10-01 10:20:00"}
{"user": 1006, "product": "Apple Watch", "order_time": "2023-10-02 08:00:00"}
{"user": 1007, "product": "Apple Watch", "order_time": "2023-10-03 08:20:00"}
{"user": 1008, "product": "Apple Watch", "order_time": "2024-10-03 08:20:00"}
{"user": 1009, "product": "Apple Watch", "order_time": "2025-10-03 08:20:00"}

3.3.2 批处理(Batch)

• 通过设置 spark.sql.sources.v2.bucketing.enabled 为 true，Spark 将识别 V2 数据源报告的特定分布，并在必要时尝试避免shuffle。
• 如下代码所示，如果两个表具有相同的分桶策略和相同数量的桶，昂贵的 join shuffle 操作将被避免。

-- 在必要时尝试避免shuffle
SET spark.sql.sources.v2.bucketing.enabled = true;

-- 事实表
CREATE TABLE FACT_TABLE (
    order_id INT, 
    f1 STRING
) TBLPROPERTIES ('bucket'='10', 'bucket-key' = 'order_id');

-- 维度表
CREATE TABLE DIM_TABLE (
    order_id INT, 
    f2 STRING
) TBLPROPERTIES ('bucket'='10', 'primary-key' = 'order_id');


SELECT 
	* 
FROM 
	FACT_TABLE 
JOIN 
	DIM_TABLE 
ON 
	FACT_TABLE.order_id = DIM_TABLE.order_id;

注：

• Paimon还有其他功能，这里就不再介绍，可以参考官网自行了解。例如：

  • Paimon 在 Flink 1.17 及后续版本中支持使用 UPDATE 更新主键表记录、使用DELETE删除change-log数据；

  • 流式读取表时指定consumer-id，防止快照因为过期而被删除；

  • paimon提供了包含有关每个表的元数据和信息的系统表，例如创建的快照和使用的选项。用户可以通过批量查询访问系统表。

    -- 快照表 Snapshots Table
    SELECT * FROM ws_t$snapshots;
    
    -- 模式表 Schemas Table
    SELECT * FROM ws_t$schemas;
    
    -- 选项表 Options Table
    SELECT * FROM ws_t$options;
    
    -- 标签表 Tags Table
    SELECT * FROM ws_t$tags;
    
    -- 审计日志表 Audit log Table
    SELECT * FROM ws_t$audit_log;
    ......
    

  • 可以集成其他引擎，如spark引擎等

  • Paimon表支持分区过期配置

  • 缩放Bucket官方示例

    • Rescale Bucket | Apache Paimon