工信部备案网站查,侧导航网站,建设银行内部审批哪些网站,wordpress主题nanaPaimon的下载及安装#xff0c;并且了解了主键表的引擎以及changelog-producer的含义参考#xff1a;
大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(1)
利用Paimon表做lookup join#xff0c;集成mysql cdc等参考#xff1a;
大数据组件(四)快速入门实时数据…Paimon的下载及安装并且了解了主键表的引擎以及changelog-producer的含义参考
大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(1)
利用Paimon表做lookup join集成mysql cdc等参考
大数据组件(四)快速入门实时数据湖存储系统Apache Paimon(2)
利用Paimon的Tag兼容HiveBranch管理等参考:
大数据组件(四)快速入门实时数据湖存储系统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, -- 延迟10mintag.num-retained-max 90 -- 最大保存90天
);上面配置表明每天0点10分钟创建一个 Tag最大保留3个月的 TagFlink 流式写入自动创建 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
);-- 插入数据
-- snapshot1
Flink SQL INSERT INTO mydb_t VALUES (1, 10, 100), (2, 20, 200);
-- snapshot2
Flink SQL INSERT INTO mydb_t VALUES (3, 30, 300), (4, 40, 400);然后利用action包创建Tag
# 利用action 包创建tag\
# 依旧利用hive元数据做catalog
[rootcentos01 ~]# /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 metastorehive \--catalog-conf urithrift://centos01:9083 \--snapshot 1[rootcentos01 ~]# /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 metastorehive \--catalog-conf urithrift://centos01:9083 \--snapshot 2 我们就可以在hive中查看分区且查询数据
0: jdbc:hive2://192.168.42.101:10000 show partitions mydb_t;
----------------
| partition |
----------------
| dt2025-02-18 |
| dt2025-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提前创建partitionsmetastore.tag-to-partition.preview process-time
);-- snapshot1
Flink SQL INSERT INTO mydb_preview VALUES (1, 10, 100), (2, 20, 200);-- create tag 2025-02-19 for snapshot 1
[rootcentos01 ~]# /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 metastorehive \--catalog-conf urithrift://centos01:9083 \--snapshot 10: jdbc:hive2://192.168.42.101:10000 show partitions mydb_preview;;
----------------
| partition |
----------------
| dt2025-02-19 |
| dt2025-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, 这个分支将负责接收实时流入的数据。
[rootcentos01 ~]# /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 metastorehive \--catalog-conf urithrift://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)即删除主表的一切数据并将分支表的一切数据拷贝到主表
[rootcentos01 ~]# /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 metastorehive \--catalog-conf urithrift://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.watermark1696126500000) */;
--------------------------------------------------------------------------
| 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 为 trueSpark 将识别 V2 数据源报告的特定分布并在必要时尝试避免shuffle。如下代码所示如果两个表具有相同的分桶策略和相同数量的桶昂贵的 join shuffle 操作将被避免。
-- 在必要时尝试避免shuffle
SET spark.sql.sources.v2.bucketing.enabled true;-- 事实表
CREATE TABLE FACT_TABLE (order_id INT, f1 STRING
) TBLPROPERTIES (bucket10, bucket-key order_id);-- 维度表
CREATE TABLE DIM_TABLE (order_id INT, f2 STRING
) TBLPROPERTIES (bucket10, 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
