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