Apache Hudi异步Compaction方式汇总

Apache Hudi异步Compaction方式汇总,第1张

Apache Hudi异步Compaction方式汇总

本篇文章对执行异步Compaction的不同部署模型一探究竟。


1. Compaction

对于Merge-On-Read表,数据使用列式Parquet文件和行式Avro文件存储,更新被记录到增量文件,然后进行同步/异步compaction生成新版本的列式文件。


Merge-On-Read表可减少数据摄入延迟,因而进行不阻塞摄入的异步Compaction很有意义。


2. 异步Compaction

异步Compaction会进行如下两个步骤

  • 调度Compaction:由摄取作业完成,在这一步,Hudi扫描分区并选出待进行compaction的FileSlice,最后CompactionPlan会写入Hudi的Timeline。


  • 执行Compaction:一个单独的进程/线程将读取CompactionPlan并对FileSlice执行Compaction *** 作。


3. 部署模型

几种执行异步Compaction的方法如下

3.1 Spark Structured Streaming

在0.6.0版本,Hudi支持在Spark Structured Streming作业中支持异步Compaction,Compactions在streaming作业内被异步调度和执行,Spark Structured作业在Merge-On-Read表中会默认开启异步Compaction。


Java代码示例如下

import org.apache.hudi.DataSourceWriteOptions;
import org.apache.hudi.HoodieDataSourceHelpers;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig; import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.ProcessingTime; DataStreamWriter<Row> writer = streamingInput.writeStream().format("org.apache.hudi")
.option(DataSourceWriteOptions.OPERATION_OPT_KEY(), operationType)
.option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY(), tableType)
.option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY(), "_row_key")
.option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY(), "partition")
.option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY(), "timestamp")
.option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_PROP, "10")
.option(DataSourceWriteOptions.ASYNC_COMPACT_ENABLE_OPT_KEY(), "true")
.option(HoodieWriteConfig.TABLE_NAME, tableName)
.option("checkpointLocation", checkpointLocation)
.outputMode(OutputMode.Append());
writer.trigger(new ProcessingTime(30000)).start(tablePath);
3.2 DeltaStreamer Continuous模式

Hudi DeltaStreamer提供连续摄入模式,Spark作业可以持续从上游消费数据写入Hudi,在该模式下,Hudi也支持异步Compaction,下面是在连续模式下进行异步Compaction示例

spark-submit --packages org.apache.hudi:hudi-utilities-bundle_2.11:0.6.0 \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer \
--table-type MERGE_ON_READ \
--target-base-path <hudi_base_path> \
--target-table <hudi_table> \
--source-class org.apache.hudi.utilities.sources.JsonDFSSource \
--source-ordering-field ts \
--schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider \
--props /path/to/source.properties \
--continous
3.3 Hudi CLI

Hudi CLI 是另一种异步执行指定Compaction的方式,示例如下

hudi:trips->compaction run --tableName <table_name> --parallelism <parallelism> --compactionInstant <InstantTime>
...
3.4 Hudi Compactor脚本

Hudi还提供了独立工具来异步执行指定Compaction,示例如下

spark-submit --packages org.apache.hudi:hudi-utilities-bundle_2.11:0.6.0 \
--class org.apache.hudi.utilities.HoodieCompactor \
--base-path <base_path> \
--table-name <table_name> \
--instant-time <compaction_instant> \
--schema-file <schema_file>
4. 总结

Hudi提供了不同的Compaction方式,可根据不同应用场景部署不同Compaction方式。


欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zaji/587186.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-04-12
下一篇 2022-04-12

发表评论

登录后才能评论

评论列表(0条)

保存