本篇文章對執行非同步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方式。