Hudi extraMetadata 研究总结

2024-06-13

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

前言

研究总结 Hudi extraMetadata ，记录研究过程。主要目的是通过 extraMetadata 保存 source 表的 commitTime （checkpoint），来实现增量读Hudi表写Hudi表时，保存增量读状态的事务性，实现类似于流任务中的 exactly-once

背景需求

有个需求：增量读Hudi表关联其他Hudi表然后做一些复杂的业务处理逻辑,然后将结果表保存到目标Hudi表中

然后就有问题：如何保证保存目标Hudi表和保存 endCommiTime 是事务的。具体如下
1、Spark支持增量读Hudi表，需要传入起（始） commitTime。
2、上一批数据处理完成后应该保存上次增量读的的 endCommiTime,作为下一批增量读的的 beginCommitTime
3、可以选择将上一批的 endCommiTime 保存到 HDFS 或者数据库表中，但是不能保证事务。（并且还需要借助外部表或者存储目录，如果能通过Hudi本身解决，就比较完美了）
4、也就是可能存在保存目标Hudi表成功，但是保存 endCommiTime 失败的情况

这时就想到了 DeltaStreamer 中是通过设置 checkpoint 来实现的，具体是通过将 commitTime 的值保存到目标表 .commit 元数据文件中(extraMetadata->deltastreamer.checkpoint.key)。我们可以参考 DeltaStreamer 中的实现逻辑，来实现我们的需求。

版本

Hudi 0.13.0
Spark 3.2.3

DeltaStreamer

官方文档：https://hudi.apache.org/docs/0.13.0/hoodie_deltastreamer/

Checkpointing

HoodieDeltaStreamer使用检查点来跟踪已经读取的数据，因此无需重新处理所有数据即可恢复。当使用Kafka源时，检查点是Kafka偏移，当使用DFS源时，该检查点是最近读取的文件的“上次修改”时间戳。检查点保存在.houdie提交文件中，作为deltastreamer.checkpoint.key。
如果需要更改检查点以重新处理或重放数据，可以使用以下选项：
–checkpoint将在提交文件中设置deltastreamer.checkpoint.reset_key以覆盖当前检查点。
–源限制将设置要从源读取的最大数据量。对于DFS源，这是读取的最大字节数。对于kafka来说，这是要阅读的事件的最大数量。

先直接看一下 DeltaStreamer 的 checkpoint 长啥样：checkpoint 保存在 .commit文件中的 extraMetadata:deltastreamer.checkpoint.key 中。

对于DFSSource，checkpoint 保存的是最近读取的文件的“上次修改”时间戳。

对于HoodieIncrSource，checkpoint 保存的Hudi表的 commitTime.

对于KafkaSource，checkpoint 保存的是 offset

示例

以下是 DFSSource 和 HoodieIncrSource 的示例，对于 DeltaStreamer 不熟悉的可以参考我之前写的文章

DFSSource

创建 source表并造数

create table test.test_source_json(
  id int,
  name string,
  price double,
  ts bigint,
  dt string
)
row format serde 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS TEXTFILE;

insert into test.test_source_json values (1,'hudi', 10.0,100,'2024-06-11');

common.properties

hoodie.datasource.write.hive_style_partitioning=true
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.ComplexKeyGenerator
hoodie.datasource.hive_sync.use_jdbc=false
hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor

json_dfs_source.properties

include=common.properties

hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.partitionpath.field=
hoodie.deltastreamer.source.dfs.root=/warehouse/tablespace/managed/hive/test.db/test_source_json

# 通过HiveSchemaProvider获取Schema
hoodie.deltastreamer.schemaprovider.source.schema.hive.database=test
hoodie.deltastreamer.schemaprovider.source.schema.hive.table=test_source_json

hoodie.datasource.hive_sync.table=test_hudi_target_json
hoodie.datasource.hive_sync.database=hudi

spark-submit --conf "spark.sql.catalogImplementation=hive" \
--principal hive/indata-10-110-105-164.indata.com@INDATA.COM --keytab /etc/security/keytabs/hive.service.keytab \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /usr/hdp/3.1.0.0-78/spark/jars/hudi-utilities-bundle_2.12-0.13.0.jar \
--props file:///opt/dkl/json_dfs_source.properties \
--target-base-path /hudi/test_hudi_target_json \
--target-table test_hudi_target_json \
--op BULK_INSERT \
--table-type COPY_ON_WRITE \
--schemaprovider-class org.apache.hudi.utilities.schema.HiveSchemaProvider \
--enable-sync

HoodieIncrSource

hudi_incr_source.properties

include=common.properties

hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.partitionpath.field=

hoodie.deltastreamer.source.hoodieincr.path=/hudi/test_hudi_target_json
hoodie.deltastreamer.source.hoodieincr.missing.checkpoint.strategy=READ_UPTO_LATEST_COMMIT
#hoodie.deltastreamer.source.hoodieincr.num_instants=1

hoodie.datasource.hive_sync.table=test_hudi_incr_target
hoodie.datasource.hive_sync.database=hudi

spark-submit --conf "spark.sql.catalogImplementation=hive" \
--principal hive/indata-10-110-105-164.indata.com@INDATA.COM --keytab /etc/security/keytabs/hive.service.keytab \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /usr/hdp/3.1.0.0-78/spark/jars/hudi-utilities-bundle_2.12-0.13.0.jar \
--source-class org.apache.hudi.utilities.sources.HoodieIncrSource \
--props file:///opt/dkl/hudi_incr_source.properties \
--target-base-path /hudi/test_hudi_incr_target \
--target-table test_hudi_incr_target \
--op BULK_INSERT \
--table-type COPY_ON_WRITE \
--enable-hive-sync

参考 DeltaStreamer

从上面的示例中，我们看到 DeltaStreamer 的 HoodieIncrSource 可以实现将 commitTime 保存到 .commit 文件中的 extraMetadata:deltastreamer.checkpoint.key 中。因为写Hudi表是事务的，这样就将写Hudi表和保存增量读状态绑定到一个事务中了。我们下次增量读的时候就可以去读目标表 .commit 文件中保存的源表的commitTime，作为增量读的beginTime，从而实现我们的需求。

但是目前还不知道 extraMetadata 是怎么保存的，所以我们需要先参考一下 DeltaStreamer 的源码，以下是部分源码：

HoodieDeltaStreamer.main -> HoodieDeltaStreamer.sync -> DeltaSync.syncOnce -> DeltaSync.syncOnce 和 DeltaSync.writeToSink
(其中 DeltaSync.syncOnce 会通过 readFromSource 获取具体的数据和checkpoint，比如 source 为 HoodieIncrSource 时，最终会调用 HoodieIncrSource.fetchNextBatch)

DeltaSync.writeToSink

private Pair<Option<String>, JavaRDD<WriteStatus>> writeToSink(JavaRDD<HoodieRecord> records, String checkpointStr,
                                                                HoodieDeltaStreamerMetrics metrics,
                                                                Timer.Context overallTimerContext) {
   ......
   // try to start a new commit
   String instantTime = startCommit();
   LOG.info("Starting commit  : " + instantTime);

   JavaRDD<WriteStatus> writeStatusRDD;
   switch (cfg.operation) {
     case INSERT:
       writeStatusRDD = writeClient.insert(records, instantTime);
       break;
     case UPSERT:
       writeStatusRDD = writeClient.upsert(records, instantTime);
       break;
     case BULK_INSERT:
       writeStatusRDD = writeClient.bulkInsert(records, instantTime);
       break;
     case INSERT_OVERWRITE:
       writeStatusRDD = writeClient.insertOverwrite(records, instantTime).getWriteStatuses();
       break;
     case INSERT_OVERWRITE_TABLE:
       writeStatusRDD = writeClient.insertOverwriteTable(records, instantTime).getWriteStatuses();
       break;
     case DELETE_PARTITION:
       List<String> partitions = records.map(record -> record.getPartitionPath()).distinct().collect();
       writeStatusRDD = writeClient.deletePartitions(partitions, instantTime).getWriteStatuses();
       break;
     default:
       throw new HoodieDeltaStreamerException("Unknown operation : " + cfg.operation);
   }

   long totalErrorRecords = writeStatusRDD.mapToDouble(WriteStatus::getTotalErrorRecords).sum().longValue();
   long totalRecords = writeStatusRDD.mapToDouble(WriteStatus::getTotalRecords).sum().longValue();
   boolean hasErrors = totalErrorRecords > 0;
   if (!hasErrors || cfg.commitOnErrors) {
     HashMap<String, String> checkpointCommitMetadata = new HashMap<>();
     if (checkpointStr != null) {
       checkpointCommitMetadata.put(CHECKPOINT_KEY, checkpointStr);
     }
     if (cfg.checkpoint != null) {
       checkpointCommitMetadata.put(CHECKPOINT_RESET_KEY, cfg.checkpoint);
     }

     if (hasErrors) {
       LOG.warn("Some records failed to be merged but forcing commit since commitOnErrors set. Errors/Total="
           + totalErrorRecords + "/" + totalRecords);
     }
     String commitActionType = CommitUtils.getCommitActionType(cfg.operation, HoodieTableType.valueOf(cfg.tableType));
     boolean success = writeClient.commit(instantTime, writeStatusRDD, Option.of(checkpointCommitMetadata), commitActionType, Collections.emptyMap());
     ……
   } else {
    ……
   }
   long overallTimeMs = overallTimerContext != null ? overallTimerContext.stop() : 0;

   // Send DeltaStreamer Metrics
   metrics.updateDeltaStreamerMetrics(overallTimeMs);
   return Pair.of(scheduledCompactionInstant, writeStatusRDD);
 }

这里writeClient 为SparkRDDWriteClient，从 writeToSink 可以看出整体逻辑：
1、startCommit()。对应生成 .requested
2、调用 writeClient 的 insert、upsert、bulkInsert等方法。对应生成 .inflight。
3、将 checkpointStr 作为 CHECKPOINT_KEY 的值放到 checkpointCommitMetadata中，通过 writeClient.commit 方法将 checkpointCommitMetadata 作为 extraMetadata 写到 .commit 文件中。对应生成 .commit

解释一下2：默认情况下 writeClient.insert 会先生成 .inflight，然后写数据，最后生成 .commit。
writeClient.insert 最终会调用 commitOnAutoCommit，因为默认情况下 config.shouldAutoCommit 是true,所以会调用 autoCommit(extraMetadata, result) 生成.commit。但是在 DeltaSync 中构造 writeClient 将 AutoCommit 配置成了false，所以步骤 2不会生成.commit，这样就可以在步骤3中手动调用 writeClient.commit 将 checkpointCommitMetadata 作为 extraMetadata 写到 .commit 文件中。

 protected void commitOnAutoCommit(HoodieWriteMetadata result) {
   // validate commit action before committing result
   runPrecommitValidators(result);
   if (config.shouldAutoCommit()) {
     LOG.info("Auto commit enabled: Committing " + instantTime);
     autoCommit(extraMetadata, result);
   } else {
     LOG.info("Auto commit disabled for " + instantTime);
   }
 }

public static final ConfigProperty<String> AUTO_COMMIT_ENABLE = ConfigProperty
     .key("hoodie.auto.commit")
     .defaultValue("true")
     .withDocumentation("Controls whether a write operation should auto commit. This can be turned off to perform inspection"
         + " of the uncommitted write before deciding to commit.");  

 private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
   final boolean combineBeforeUpsert = true;
   // 这里将 autoCommit 配置为 false。
   final boolean autoCommit = false;

   // NOTE: Provided that we're injecting combined properties
   //       (from {@code props}, including CLI overrides), there's no
   //       need to explicitly set up some configuration aspects that
   //       are based on these (for ex Clustering configuration)
   HoodieWriteConfig.Builder builder =
       HoodieWriteConfig.newBuilder()
           .withPath(cfg.targetBasePath)
           .combineInput(cfg.filterDupes, combineBeforeUpsert)
           .withCompactionConfig(
               HoodieCompactionConfig.newBuilder()
                   .withInlineCompaction(cfg.isInlineCompactionEnabled())
                   .build()
           )
           .withPayloadConfig(
               HoodiePayloadConfig.newBuilder()
                   .withPayloadClass(cfg.payloadClassName)
                   .withPayloadOrderingField(cfg.sourceOrderingField)
                   .build())
           .forTable(cfg.targetTableName)
           .withAutoCommit(autoCommit)
           .withProps(props);

从上面的分析中，我们知道了 extraMetadata 是通过 writeClient.commit 方法写到.commit 文件中的，并不是通过参数配置实现的。这属于Hudi源码中比较底层的代码，如果是 Hudi Java Client 可以参考这样实现，但是 Spark Client 是封装好的，这样实现并不通用。最好是研究一下有没有通过参数配置实现的这种比较简单的方案，如果实在没有再考虑自己参考 DeltaStreamer 的逻辑实现，封装一个公用的实现方法，或者修改源码，所以接下来先看一下 Spark Client 的源码：HoodieSparkSqlWriter

HoodieSparkSqlWriter

HoodieSparkSqlWriter.write

def write(sqlContext: SQLContext,
          mode: SaveMode,
          optParams: Map[String, String],
          df: DataFrame,
          hoodieTableConfigOpt: Option[HoodieTableConfig] = Option.empty,
          hoodieWriteClient: Option[SparkRDDWriteClient[_]] = Option.empty,
          asyncCompactionTriggerFn: Option[SparkRDDWriteClient[_] => Unit] = Option.empty,
          asyncClusteringTriggerFn: Option[SparkRDDWriteClient[_] => Unit] = Option.empty,
          extraPreCommitFn: Option[BiConsumer[HoodieTableMetaClient, HoodieCommitMetadata]] = Option.empty):
(Boolean, HOption[String], HOption[String], HOption[String], SparkRDDWriteClient[_], HoodieTableConfig) = {
    ……
    val (writeResult, writeClient: SparkRDDWriteClient[_]) =
      operation match {
        case WriteOperationType.DELETE =>
          ……
          // Issue deletes
          client.startCommitWithTime(instantTime, commitActionType)
          val writeStatuses = DataSourceUtils.doDeleteOperation(client, hoodieKeysToDelete, instantTime)
          (writeStatuses, client)

        case WriteOperationType.DELETE_PARTITION =>
          ……
          client.startCommitWithTime(instantTime, commitActionType)
          val writeStatuses = DataSourceUtils.doDeletePartitionsOperation(client, partitionsToDelete, instantTime)
          (writeStatuses, client)

        // Here all other (than DELETE, DELETE_PARTITION) write operations are handled
        case _ =>
          ……
          val client = hoodieWriteClient.getOrElse {
            val finalOpts = addSchemaEvolutionParameters(parameters, internalSchemaOpt, Some(writerSchema)) - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key
            // TODO(HUDI-4772) proper writer-schema has to be specified here
            DataSourceUtils.createHoodieClient(jsc, dataFileSchema.toString, path, tblName, mapAsJavaMap(finalOpts))
          }
          val writeConfig = 
          client.startCommitWithTime(instantTime, commitActionType)
          val writeResult = DataSourceUtils.doWriteOperation(client, dedupedHoodieRecords, instantTime, operation)
          (writeResult, client)
      }

    // Check for errors and commit the write.
    try {
      val (writeSuccessful, compactionInstant, clusteringInstant) =
        commitAndPerformPostOperations(sqlContext.sparkSession, df.schema,
          writeResult, parameters, writeClient, tableConfig, jsc,
          TableInstantInfo(basePath, instantTime, commitActionType, operation), extraPreCommitFn)

      (writeSuccessful, common.util.Option.ofNullable(instantTime), compactionInstant, clusteringInstant, writeClient, tableConfig)
    } finally {
      // close the write client in all cases
      val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient, tableConfig, parameters, jsc.hadoopConfiguration())
      val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient, parameters)
      if (!asyncCompactionEnabled && !asyncClusteringEnabled) {
        log.info("Closing write client")
        writeClient.close()
      }
    }
  }
}

和 DeltaStreamer 相同的逻辑，分三步：
1、client.startCommitWithTime 。对应生成 .requested
2、DataSourceUtils.doWriteOperation ，其中这里的client 是通过 DataSourceUtils.createHoodieClient 创建的，也设置了 autoCommit 为 false。对应生成 .inflight。
3、commitAndPerformPostOperations 。对应生成 .commit

接下来看一下 commitAndPerformPostOperations 有没有可以通过配置参数来设置 extraMetadata的实现方法。

private def commitAndPerformPostOperations(spark: SparkSession,
                                           schema: StructType,
                                           writeResult: HoodieWriteResult,
                                           parameters: Map[String, String],
                                           client: SparkRDDWriteClient[_],
                                           tableConfig: HoodieTableConfig,
                                           jsc: JavaSparkContext,
                                           tableInstantInfo: TableInstantInfo,
                                           extraPreCommitFn: Option[BiConsumer[HoodieTableMetaClient, HoodieCommitMetadata]]
                                          ): (Boolean, common.util.Option[java.lang.String], common.util.Option[java.lang.String]) = {
  if (writeResult.getWriteStatuses.rdd.filter(ws => ws.hasErrors).count() == 0) {
    log.info("Proceeding to commit the write.")
    val metaMap = parameters.filter(kv =>
      kv._1.startsWith(parameters(COMMIT_METADATA_KEYPREFIX.key)))
    val commitSuccess =
      client.commit(tableInstantInfo.instantTime, writeResult.getWriteStatuses,
        common.util.Option.of(new java.util.HashMap[String, String](mapAsJavaMap(metaMap))),
        tableInstantInfo.commitActionType,
        writeResult.getPartitionToReplaceFileIds,
        common.util.Option.ofNullable(extraPreCommitFn.orNull))

    ……
}

这里惊喜的发现有一个有可能实现我们需求的参数：COMMIT_METADATA_KEYPREFIX.key

COMMIT_METADATA_KEYPREFIX

val COMMIT_METADATA_KEYPREFIX: ConfigProperty[String] = ConfigProperty
  .key("hoodie.datasource.write.commitmeta.key.prefix")
  .defaultValue("_")
  .withDocumentation("Option keys beginning with this prefix, are automatically added to the commit/deltacommit metadata. " +
    "This is useful to store checkpointing information, in a consistent way with the hudi timeline")

以该前缀开头的选项键会自动添加到commit/detacommit元数据中。这对于以与hudi时间线一致的方式存储检查点信息非常有用.

根据文档注释发现该参数确实有可能能实现我们的需求，但是该参数该怎么用呢？网上并没有找到相关的文档，简单尝试一下，发现也并不生效。最终参考源码中的测试用例实现：

@Test
public void testDataSourceWriterExtraCommitMetadata() throws Exception {
  String commitExtraMetaPrefix = "commit_extra_meta_";
  Map<String, String> extraMeta = new HashMap<>();
  extraMeta.put(DataSourceWriteOptions.COMMIT_METADATA_KEYPREFIX().key(), commitExtraMetaPrefix);
  extraMeta.put(commitExtraMetaPrefix + "a", "valA");
  extraMeta.put(commitExtraMetaPrefix + "b", "valB");
  extraMeta.put("commit_extra_c", "valC"); // should not be part of commit extra metadata

  Map<String, String> expectedMetadata = new HashMap<>();
  expectedMetadata.putAll(extraMeta);
  expectedMetadata.remove(DataSourceWriteOptions.COMMIT_METADATA_KEYPREFIX().key());
  expectedMetadata.remove("commit_extra_c");

  testDataSourceWriterInternal(extraMeta, expectedMetadata, true);
}

写示例

根据测试用例中的用法，简单尝试一下，看是否生效

create table hudi.test_hudi_table (
  id int,
  name string,
  price double,
  ts long,
  dt string
) using hudi
 partitioned by (dt)
 options (
  type = 'cow',
  primaryKey = 'id',
  preCombineField = 'ts',
  hoodie.datasource.write.commitmeta.key.prefix='commit_extra_meta_',
  commit_extra_meta_a='valA',
  commit_extra_meta_b='valB'
 ); 

insert into hudi.test_hudi_table values (1,'hudi',10,100,'2024-06-11');

结果发现成功添加了 extraMetadata ，其中 schema 是默认自带的，commit_extra_meta_b 和 commit_extra_meta_a 是新添加的。

"extraMetadata" : {
  "schema" : "{\"type\":\"record\",\"name\":\"test_hudi_table_record\",\"namespace\":\"hoodie.test_hudi_table\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"name\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"price\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"ts\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"dt\",\"type\":[\"null\",\"string\"],\"default\":null}]}",
  "commit_extra_meta_b" : "valB",
  "commit_extra_meta_a" : "valA"
},

但是这种方式，在建表语句里写死了值，不能灵活设置。我们的需求是要灵活设置，因为每次insert 对应的 beginCommitTime 都会变化，我们尝试验证一下能不能灵活配置。

SQL SET

create table hudi.test_hudi_table (
  id int,
  name string,
  price double,
  ts long,
  dt string
) using hudi
 partitioned by (dt)
 options (
  type = 'cow',
  primaryKey = 'id',
  preCombineField = 'ts'
 ); 

set hoodie.datasource.write.commitmeta.key.prefix=commit_extra_meta_;

set commit_extra_meta_a='valA';
set commit_extra_meta_b='valB';  
insert into hudi.test_hudi_table values (1,'hudi',10,100,'2024-06-11');

set commit_extra_meta_a='new_valA';
set commit_extra_meta_b='new_valB';
insert into hudi.test_hudi_table values (2,'hudi',20,200,'2024-06-11');

经验证，不生效。原因是Hudi Spark SQL 源码里过滤掉了不是以 hoodie. 开头的配置项。也就是要求配置项必须以 hoodie. 开头才生效，这样我们设置的 commit_extra_meta_a 和 commit_extra_meta_b 就不会生效。
相关源码：

InsertIntoHoodieTableCommand.run -> ProvidesHoodieConfig.buildHoodieInsertConfig

def run(sparkSession: SparkSession,
        table: CatalogTable,
        query: LogicalPlan,
        partitionSpec: Map[String, Option[String]],
        overwrite: Boolean,
        refreshTable: Boolean = true,
        extraOptions: Map[String, String] = Map.empty): Boolean = {
  val catalogTable = new HoodieCatalogTable(sparkSession, table)
  ……

  val config = buildHoodieInsertConfig(catalogTable, sparkSession, isOverWritePartition, isOverWriteTable, partitionSpec, extraOptions)
  ……

  val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sparkSession.sqlContext, mode, config, Dataset.ofRows(sparkSession, alignedQuery))

  ……
  success
}

def buildHoodieInsertConfig(hoodieCatalogTable: HoodieCatalogTable,
                            sparkSession: SparkSession,
                            isOverwritePartition: Boolean,
                            isOverwriteTable: Boolean,
                            insertPartitions: Map[String, Option[String]] = Map.empty,
                            extraOptions: Map[String, String]): Map[String, String] = {

  ……

  combineOptions(hoodieCatalogTable, tableConfig, sparkSession.sqlContext.conf,
    defaultOpts = defaultOpts, overridingOpts = overridingOpts)
}

def combineOptions(catalogTable: HoodieCatalogTable,
                           tableConfig: HoodieTableConfig,
                           sqlConf: SQLConf,
                           defaultOpts: Map[String, String],
                           overridingOpts: Map[String, String] = Map.empty): Map[String, String] = {

  filterNullValues(defaultOpts) ++
    DFSPropertiesConfiguration.getGlobalProps.asScala.toMap ++
    mapSqlOptionsToDataSourceWriteConfigs(catalogTable.catalogProperties) ++
    tableConfig.getProps.asScala.toMap ++
    filterHoodieConfigs(sqlConf.getAllConfs) ++
    filterNullValues(overridingOpts)
}

private def filterHoodieConfigs(opts: Map[String, String]): Map[String, String] =
  opts.filterKeys(isHoodieConfigKey)

def isHoodieConfigKey(key: String): Boolean =
  key.startsWith("hoodie.") || key == DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key

所以我们可以将 prefix 改为以 hoodie. 开头：

set hoodie.datasource.write.commitmeta.key.prefix=hoodie.commit_extra_meta_;

set hoodie.commit_extra_meta_a='valA';
set hoodie.commit_extra_meta_b='valB';  
insert into hudi.test_hudi_table values (1,'hudi',10,100,'2024-06-11');

set hoodie.commit_extra_meta_a='new_valA';
set hoodie.commit_extra_meta_b='new_valB';
insert into hudi.test_hudi_table values (2,'hudi',20,200,'2024-06-11');

经验证，生效：

但是这样限制并不合理，比如 DeltaStreamer 的 key 为 deltastreamer.checkpoint.key ，就不是以 hoodie. 开头，假如我们有这样的需求，不想被限制住，就可以通过修改源码来解决。
源码：https://github.com/dongkelun/hudi/commit/f8c3d547928af0d3287de42e3d0148970f21472c
jar包：https://download.csdn.net/download/dkl12/89429544

注意：set prefix 时不要加引号：

1 2	## 加引号是有问题的 set hoodie.datasource.write.commitmeta.key.prefix='hoodie.commit_extra_meta_';

原因：调试代码发现，解析时没有将引号删掉，导致过滤失效。’hoodie.commit_extrameta‘ 不等于 hoodie.commit_extrameta 。这样代码 val metaMap = parameters.filter(kv => kv._1.startsWith(parameters(COMMIT_METADATA_KEYPREFIX.key))) 返回空。

代码

根据我对Hudi源码的理解：因为代码和SQL的入口不一样，代码可能并没有 hoodie. 开头的限制。接下来验证一下代码有没有 hoodie. 开头的限制，会不会生效。（先将包还原）

import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.config.HoodieWriteConfig.TBL_NAME
import org.apache.hudi.hive.MultiPartKeysValueExtractor
import org.apache.hudi.keygen.ComplexKeyGenerator
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

def save2HudiSyncHive(df: DataFrame, databaseName: String, tableName: String, primaryKey: String, preCombineField: String,
                                    partitionField: String, operation: String, tablePath: String, metaA: String, metaB: String): Unit = {
  df.
    write.format("hudi").
    option(RECORDKEY_FIELD.key, primaryKey).
    option(PRECOMBINE_FIELD.key, preCombineField).
    option(PARTITIONPATH_FIELD.key, partitionField).
    option(TBL_NAME.key, tableName).
    option(KEYGENERATOR_CLASS_NAME.key(), classOf[ComplexKeyGenerator].getName).
    option(OPERATION.key(), operation).
    option(META_SYNC_ENABLED.key, true).
    option(HIVE_USE_JDBC.key, false).
    option(HIVE_DATABASE.key, databaseName).
    option(HIVE_AUTO_CREATE_DATABASE.key, true).
    option(HIVE_CREATE_MANAGED_TABLE.key, true).
    option(HIVE_TABLE.key, tableName).
    option(HIVE_CREATE_MANAGED_TABLE.key, true).
    option(HIVE_STYLE_PARTITIONING.key, true).
    option(HIVE_PARTITION_FIELDS.key, partitionField).
    option(HIVE_PARTITION_EXTRACTOR_CLASS.key, classOf[MultiPartKeysValueExtractor].getName).
    option("hoodie.datasource.write.commitmeta.key.prefix", "commit_extra_meta_").
    option("commit_extra_meta_a", metaA).
    option("commit_extra_meta_b", metaB).
    mode("Append").
    save(tablePath)
}

import spark.implicits._
val df = Seq((1, "a1", 10, 100, "2024-06-11")).toDF("id", "name", "value", "ts", "dt")
val databaseName = "hudi"
val tableName = "test_hudi_table3"
val primaryKey = "id"
val preCombineField = "ts"
val partitionField = "dt"
val tablePath = "/tmp/hudi/test_hudi_table3"

save2HudiSyncHive(df, databaseName, tableName, primaryKey, preCombineField, partitionField,
  UPSERT_OPERATION_OPT_VAL, tablePath, "valA", "valB" )
save2HudiSyncHive(df, databaseName, tableName, primaryKey, preCombineField, partitionField,
  UPSERT_OPERATION_OPT_VAL, tablePath, "new_valA", "new_valA" )

验证结果：确实没有限制，所以对于没有必须使用SQL的需求，可以不用修改源码，直接写代码就可以搞定。比如我本次的需求，因为业务逻辑复杂，只能写代码实现，这样即使不修改源码，也可以实现我的需求了。

读

前面我们已经成功的写入了 extraMetadata，该怎么获取它的值来使用呢？可以使用 SQL 也可以使用代码获取。本文仅展示简单的使用 show_commit_extra_metadata 来获取 extraMetadata 的值，关于代码实现获取 extraMetadata 以及实现本文开头提及的需求，会在后面的文章单独总结。

spark-sql> call show_commit_extra_metadata(table => 'hudi.test_hudi_table2', metadata_key => 'commit_extra_meta_a');
20240613143257097       commit  commit_extra_meta_a     'new_valA'
Time taken: 0.173 seconds, Fetched 1 row(s)
spark-sql> call show_commit_extra_metadata(table => 'hudi.test_hudi_table3', metadata_key => 'commit_extra_meta_a');
20240613154013674       commit  commit_extra_meta_a     new_valA
Time taken: 0.143 seconds, Fetched 1 row(s)

本文由 董可伦 发表于伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Hudi extraMetadata 研究总结

本文链接：https://dongkelun.com/2024/06/13/hudi/hudiExtraMetadata/

欢迎关注我的公众号