笑话网站html模板,上海市工程建设监理协会网站,怎么做旅游网站推广,建设银行网站不能登录文章目录 1. 引言2. 基本概念2.1 定义2.2 使用场景 3. 流式处理3.1 自动小文件合并3.2 流式查询 4. 数据更新4.1 查询4.2 更新4.3 分桶附加表 5 总结 1. 引言
通过本文#xff0c;上篇我们了解了Apache Paimon 主键表#xff0c;本期我们将继续学习附加表#xff08;Append… 文章目录 1. 引言2. 基本概念2.1 定义2.2 使用场景 3. 流式处理3.1 自动小文件合并3.2 流式查询 4. 数据更新4.1 查询4.2 更新4.3 分桶附加表 5 总结 1. 引言
通过本文上篇我们了解了Apache Paimon 主键表本期我们将继续学习附加表Append Only Table 我们将带领读者《 《Apache Paimon Docs - Table w/o PK》》 继续剖析 Paimon 的仅追加表相关知识。
通过本文你将了解到
Paimon 附加表相关的基本概念了解什么是附加表它在Paimon中扮演什么角色以及它如何与主键表区分开来。及其适用场景探索附加表在实际应用中的多样化场景。数据查询更新方式从高效的数据合并策略到灵活的流式查询配置以及如何通过索引和文件索引优化查询性能。
2. 基本概念
2.1 定义
如果一个表没有定义主键那它就是一个附加表Append Table。与主键表相比附加表无法直接接收变更日志也不能直接通过 upsert 更新数据只能接收附加数据。
CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
) WITH (-- target-file-size 256 MB,-- file.format parquet,-- file.compression zstd,-- file.compression.zstd-level 3
);2.2 使用场景
使用场景或优势说明批量写入和批量读取类似于常规的 Hive 分区表适用于大规模数据的批量处理。友好的对象存储良好的兼容性和适应性支持 S3、OSS 等对象存储。时间穿越和回滚支持数据的时间旅行和回滚功能方便数据的历史查询和恢复。低成本的删除和更新在批量数据操作中能够以较低的计算和资源成本进行删除和更新操作。流式接收中的小文件自动合并在流式写入过程中自动处理小文件合并减少存储碎片。队列形式的流式读写支持如队列般的流式读写操作可以像消息队列一样处理数据。高性能查询通过顺序和索引实现的高效查询性能。
批量写入和读取
CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
) WITH (target-file-size 256 MB, -- 设置目标文件大小file.format parquet, -- 文件格式为 Parquetfile.compression zstd, -- 使用 ZSTD 压缩file.compression.zstd-level 3 -- 设置 ZSTD 压缩级别为 3
);流式接收和小文件合并
CREATE TABLE my_stream_table (event_id BIGINT,event_time TIMESTAMP,event_data STRING
) WITH (target-file-size 128 MB, -- 设置目标文件大小file.format avro, -- 文件格式为 Avrofile.compression snappy, -- 使用 Snappy 压缩streaming.min-batch-interval 5 min -- 设置流处理最小批处理时间间隔为 5 分钟
);具有以下的优点
功能特性技术优势实现对象存储友好良好的兼容性和适应性支持 S3、OSS 等对象存储。通过接入主流对象存储服务优化读写性能和兼容性特别是大规模数据存储和处理场景下。时间穿越和回滚支持数据的时间旅行和回滚功能方便数据的历史查询和恢复。利用快照和元数据管理实现任意时间点的数据查询和回滚能力。低成本的删除和更新在批量数据操作中能够以较低的计算和资源成本进行删除和更新操作。通过高效的数据合并和变更处理机制优化批量操作中的资源消耗。小文件合并在流式写入过程中自动处理小文件合并减少存储碎片。在流式写入过程中使用异步任务定期合并小文件确保合理的文件大小和存储效率。高性能查询通过顺序和索引实现的高效查询性能。通过索引构建和数据排序提升查询的响应速度和资源利用效率。
3. 流式处理
附加表Append Table可以通过 Flink 进行非常灵活的流式写入并可以像队列一样通过 Flink 进行读取。唯一的区别是其延迟为分钟级别但其优势在于非常低的成本以及能够进行过滤和投影下推。
3.1 自动小文件合并 在流式写入作业中如果没有定义分桶bucket写入器不会进行压缩相反将使用压缩协调器Compact Coordinator扫描小文件并将压缩任务传递给压缩工作者Compact Worker。流式模式下如果在 Flink 中运行插入 SQL拓扑结构将如下所示
Source - Transformations - Sink- Compact Coordinator - Compact Worker无反压压缩任务不会引起反压。写入模式如果设置 write-only 为 true压缩协调器和压缩工作者将在拓扑中被移除。Flink 流模式自动压缩仅在 Flink 引擎的流模式下被支持。可以通过 Paimon 在 Flink 中启动压缩作业并通过设置 write-only 禁用所有其他压缩。
3.2 流式查询
附加表可以像消息队列一样使用进行流式查询与主键表类似有两个选项可以进行流式读取
默认模式流式读取在首次启动时生成表的最新快照并继续读取最新的增量记录。增量模式可以指定 scan.mode 或 scan.snapshot-id 或 scan.timestamp-millis 或 scan.file-creation-time-millis 进行增量读取。
类似 Flink-Kafka默认情况下不保证顺序。如果数据需要某种顺序也需要考虑定义桶键bucket-key请参考分桶附加Bucketed Append部分。
流式写入和自动小文件合并
CREATE TABLE my_stream_table (event_id BIGINT,event_time TIMESTAMP,event_data STRING
) WITH (target-file-size 128 MB, -- 设置目标文件大小file.format avro, -- 文件格式为 Avrofile.compression snappy, -- 使用 Snappy 压缩streaming.min-batch-interval 5 min -- 设置流处理最小批处理时间间隔为 5 分钟
);在流式写入过程中配置 Compact Coordinator 和 Compact Worker 以确保小文件自动合并。
流式查询配置默认模式
SET scan.startup.mode latest-offset; -- 设置流式读取从最新的快照开始流式查询配置增量模式
SET scan.mode incremental; -- 设置流式读取为增量模式
SET scan.snapshot-id 1234567890; -- 可选指定从特定快照 ID 开始
SET scan.timestamp-millis 1627849923000; -- 可选指定从特定时间戳毫秒开始流式查询配置带顺序要求
CREATE TABLE ordered_stream_table (event_id BIGINT,event_time TIMESTAMP,event_data STRING
) WITH (target-file-size 128 MB,file.format parquet,file.compression zstd,streaming.min-batch-interval 5 min,bucket-key event_time -- 设置桶键bucket-key以确保数据按照时间顺序
);技术优势及其实现
功能特性技术优势实现流式写入通过灵活的配置选项实现分钟级别低延迟的流式写入并支持过滤和投影下推提升查询效率。通过灵活的配置选项优化数据流的写入路径减少延迟并通过下推操作提升查询效率。自动小文件合并在流式处理过程中动态管理文件大小减少存储碎片提高存储效率。使用动态文件管理策略自动合并小文件以优化存储空间和提高I/O效率。流式读取支持从最新快照读取或增量读取类似消息队列的使用方便实时数据处理和分析。提供快照和增量读取功能使得流式读取更加灵活适用于实时数据处理场景。顺序保证通过配置桶键可以确保数据在需要顺序的情境下有序读取和写入满足业务需求。通过桶键配置实现数据的有序存储和检索保证业务逻辑的顺序性。
4. 数据更新
4.1 查询
按顺序跳过数据
Paimon 默认在清单文件中记录每个字段的最大值和最小值。在查询时根据查询的 WHERE 条件通过清单中的统计信息进行文件过滤。如果过滤效果良好查询时间可以从分钟级别加速到毫秒级别。
然而数据分布并不总是能有效过滤因此如果可以根据 WHERE 条件中的字段对数据进行排序将会更高效。可以参考 Flink 的 COMPACT Action 或 COMPACT Procedure以及 Spark 的 COMPACT Procedure。
-- 对数据进行排序以优化按顺序跳过数据的查询性能
ALTER TABLE my_table COMPACT BY (field_name);按文件索引跳过数据
还可以使用文件索引它将在读取端通过索引过滤文件。
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 后Paimon 将为每个文件创建相应的索引文件。如果索引文件太小它将直接存储在清单中否则将存储在数据文件的目录中。每个数据文件对应一个索引文件该文件有独立的定义可以包含不同类型的多列索引。
文件索引的应用场景
不同文件索引在不同场景下效率不同。例如
布隆过滤器Bloom Filter在点查找场景中可能加速查询。位图Bitmap可能消耗更多空间但精度更高。
目前文件索引仅支持附加表Append-Only Table。
布隆过滤器的配置
file-index.bloom-filter.columns指定需要布隆过滤器索引的列。file-index.bloom-filter.column_name.fpp配置错误正率False Positive Probability。file-index.bloom-filter.column_name.items配置一个数据文件中预期的不同项目数量。
位图的配置
file-index.bitmap.columns指定需要位图索引的列。
添加文件索引到现有表
如果你想在不重写的情况下添加文件索引可以使用 rewrite_file_index 过程。在使用该过程之前你应该在目标表中配置适当的配置。可以使用 ALTER 子句来配置 file-index.filter-type.columns。
使用示例添加文件索引到现有表
ALTER TABLE my_table
SET (file-index.bloom-filter.columns product_id);CALL rewrite_file_index(my_table);4.2 更新
目前仅 Spark SQL 支持 DELETE 和 UPDATE 操作可以参考 Spark Write 的相关文档。
DELETE FROM my_table
WHERE currency UNKNOWN;更新模式
附加表Append Table有两种更新模式 COWCopy on Write 机制搜索命中的文件然后重新写入每个文件以移除需要删除的数据。成本这种操作成本高因为每次删除或更新都需要重新写入整个文件。 MOWMerge on Write 机制通过指定 deletion-vectors.enabled true启用删除向量模式Deletion Vectors。只标记对应文件的某些记录为删除并写入删除文件而不需要重新写入整个文件。优势相比 COW 模式MOW 模式的删除和更新成本更低因为只需写入小的删除文件而不需要重写全部数据文件。
在创建或更新表时可以启用删除向量模式
CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
) WITH (deletion-vectors.enabled true
);或在现有表上启用删除向量
ALTER TABLE my_table
SET (deletion-vectors.enabled true);MOW 模式下的 DELETE 操作
DELETE FROM my_table
WHERE currency UNKNOWN;此操作将标记 currency 为 ‘UNKNOWN’ 的记录为删除而不重写整个文件。
通过 Spark SQL 进行更新操作
val spark SparkSession.builder().appName(UpdateExample).getOrCreate()// 启用删除向量
spark.sql(ALTER TABLE my_table SET (deletion-vectors.enabled true))// 执行 DELETE 操作
spark.sql(DELETE FROM my_table WHERE currency UNKNOWN)// 执行 UPDATE 操作
spark.sql(UPDATE my_table SET price price * 1.1 WHERE product_id 1001)4.3 分桶附加表
您可以定义 bucket 和 bucket-key 以创建一个分桶附加表。在这种表中不同桶内的数据是严格有序的流式读取将按写入顺序准确地传输记录。这样可以优化数据处理和查询性能。
--创建分桶附加表
CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
) WITH (bucket 8,bucket-key product_id
);一个普通的附加表的流式写读取没有严格的顺序保证但是有些情况下需要定义一个类似于 Kafka 的键。
每个分桶中的记录都是严格有序的流式读取将按写入顺序准确地传输记录。无需配置特殊的设置所有数据将按队列形式进入一个桶内。
分桶中的压缩Compaction in Bucket
默认情况下Sink 节点会自动执行压缩以控制文件数量。以下选项控制压缩策略
KeyDefaultTypeDescriptionwrite-onlyfalseBoolean如果设置为 true将跳过压缩和快照过期操作。此选项与专用压缩作业一起使用。compaction.min.file-num5Integer对文件集 [f_0,…,f_N]满足 sum(size(f_i)) targetFileSize 的最小文件数量以触发附加表的压缩。避免压缩几乎满的文件因为这不合算。compaction.max.file-num5Integer对文件集 [f_0,…,f_N]即使 sum(size(f_i)) targetFileSize也触发压缩的最大文件数量。此值避免过多小文件积压减慢性能。full-compaction.delta-commits(none)Integer在 delta 提交后会不断触发全量压缩。
流式读取顺序Streaming Read Order
对于流式读取记录按以下顺序生产
跨分区记录如果 scan.plan-sort-partition 设置为 true则首先生产分区值较小的记录。否则先生产创建时间较早的分区的记录。同分区同桶记录首先生产先写入的记录。同分区不同桶记录不同桶由不同任务处理不保证顺序。
水印定义Watermark Definition
CREATE TABLE t (user BIGINT,product STRING,order_time TIMESTAMP(3),WATERMARK FOR order_time AS order_time - INTERVAL 5 SECOND
) WITH (...);启动有界流任务读取 Paimon 表
SELECT window_start, window_end, COUNT(user)
FROM TABLE(TUMBLE(TABLE t, DESCRIPTOR(order_time), INTERVAL 10 MINUTES)
)
GROUP BY window_start, window_end;还可以启用 Flink 水印对齐确保没有来源/分片/分区的水印前进太快
KeyDefaultTypeDescriptionscan.watermark.alignment.group(none)String要对齐水印的一组源。scan.watermark.alignment.max-drift(none)Duration对齐水印的最大漂移在此漂移前暂停从源/任务/分区消费。
有界流Bounded Stream
流式来源Streaming Source也可以是有界的可以通过指定 scan.bounded.watermark 来定义有界流模式的结束条件。
--创建 Kafka 表和启动流式插入及读取作业
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 kafka_table;-- 启动有界流任务读取 Paimon 表
SELECT * FROM paimon_table /* OPTIONS(scan.bounded.watermark...) */;批处理Batch
分桶表可以在批处理查询中避免 shuffle例如可以用以下 Spark SQL 读取 Paimon 表
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;通过设置 spark.sql.sources.v2.bucketing.enabled 为 trueSpark 将识别 V2 数据源报告的特定分布并在必要时尝试避免 shuffle。如果两个表具有相同的分桶策略和相同数量的桶昂贵的 join shuffle 操作将被避免。
5 总结
本文详细介绍了Apache Paimon中附加表的概念和应用。我们首先定义了什么是附加表并比较了它与主键表的区别。接着我们探讨了附加表在不同场景下的使用包括批量写入和读取、对象存储的友好性、时间穿越和回滚功能、低成本的删除和更新操作、流式接收中小文件的自动合并、队列形式的流式读写以及高性能查询。此外我们还详细介绍了流式处理的相关技术包括自动小文件合并、流式查询的不同模式、顺序保证的重要性以及分桶附加表的优势。最后我们讨论了数据更新策略包括DELETE和UPDATE操作以及如何通过配置优化查询性能。
如果你想参与讨论请 点击这里https://github.com/hiszm/BigDataWeekly每周都有新的主题周末或周一发布。
大数据精读探索知识的深度。 关注 大数据精读周刊 版权声明自由转载-非商用-非衍生-保持署名[创意共享 3.0 许可证](https://creativecommons.org/licenses/by-nc-nd/3.0/deed.e
