背景
目前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
