Apache Hudi初探(六)(与spark的结合)

背景

目前hudi的与spark的集合还是基于spark datasource V1来的，这一点可以查看hudi的source实现就可以知道:

class DefaultSource extends RelationProvider
  with SchemaRelationProvider
  with CreatableRelationProvider
  with DataSourceRegister
  with StreamSinkProvider
  with StreamSourceProvider
  with SparkAdapterSupport
  with Serializable {

后续发现也是基于Datasource V2的

闲说杂谈

继续上次的Apache Hudi初探(五)涉及的代码：

  preCommit(inflightInstant, metadata);
  commit(table, commitActionType, instantTime, metadata, stats);
  // already within lock, and so no lock requried for archival
  postCommit(table, metadata, instantTime, extraMetadata, false);
  LOG.info("Committed " + instantTime);
  releaseResources();
  this.txnManager.endTransaction(Option.of(inflightInstant));
  ...
  runTableServicesInline(table, metadata, extraMetadata)
  emitCommitMetrics(instantTime, metadata, commitActionType);
    // callback if needed.
    if (config.writeCommitCallbackOn()) {
      if (null == commitCallback) {
        commitCallback = HoodieCommitCallbackFactory.create(config);
      }
      commitCallback.call(new HoodieWriteCommitCallbackMessage(instantTime, config.getTableName(), config.getBasePath(), stats));
    }
    return true;
  }

之前说到只有获取到分布式锁的线程才可以继续一下操作

preCommit 解决潜在存在的写元数据的文件冲突，因为有可能当前的写入和后台Compaction/Clustering操作存在冲突

commit 真正的写元数据的

• 在写元数据之前得了解一下marker文件 ，marker文件的创建是HoodieRowCreateHandle类中：

  createMarkerFile(partitionPath, fileName, instantTime, table, writeConfig);

这里会根据hoodie.write.markers.type的值（默认是"TIMELINE_SERVER_BASED"）以及hoodie.embed.timeline.server（默认是true）

来创建marker文件，对于默认值来说（也就是TimelineServerBasedWriteMarkers如果是基于HDFS的，就是DirectWriteMarkers），

就会向内置的TimelineServerBasedWriteMarkers服务发送创建marker文件的请求（文件的后缀为*.parquet.marker.CREATE*）,

而该服务的创建是在BaseHoodieClient的构造方法中：

 startEmbeddedServerView()

在commit文件的时候，会调用reconcileAgainstMarkers，会删除由于spark speculation导致的部分文件，这也是marker文件存在的意义

并且更新指标如"duration"和“numFilesFinalized”

writeTableMetadata 更新table的元数据

activeTimeline.saveAsComplete 创建类似20230422183552014.deltacommit的元数据文件

postCommit做一些commit后的清理工作

quietDeleteMarkerDir 删除对应的marker文件

autoCleanOnCommit 如果hoodie.clean.automatic为true且hoodie.clean.async 为false的情况下，就进行同步清理

主要是清理不需要的历史文件（parquet数据文件或者log文件）

autoArchiveOnCommit如果hoodie.archive.automatic为true且hoodie.archive.async为false，就进行归档操作

主要是对instant归档，减少timeline的操作压力

• releaseResources 在这里主要是释放掉持久化的RDD
• 
  
• this.txnManager.endTransaction(Option.of(inflightInstant))
  结束事务，释放掉锁

runTableServicesInline 这个是在事务之外的，因为这个操作是个耗时的操作，包含了Compaction和Clustering

如果 hoodie.metadata.enable 为true(默认就是true) 就会继续进行后续操作

如果 hoodie.compact.inline为false（默认就是false）并且 hoodie.compact.schedule.inline 为true（默认是false）就是会进行Compaction

如果 hoodie.compact.inline 为true,也是会进行Compaction

Clustering的操作也一样，只不过对应的配合为hoodie.clustering.inline和hoodie.clustering.schedule.inline

emitCommitMetrics指标进行采集

如果hoodie.write.commit.callback.on为true(默认是false)

还会回调hoodie.write.commit.callback.class的类（默认为org.apache.hudi.callback.impl.HoodieWriteCommitHttpCallback），根据发送一些信息给hoodie.write.commit.callback.http.url配置的http server