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方式。