欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Hudi剖析|Apache Hudi Rollback实现分析

程序员文章站 2022-07-14 20:25:37
...

1. 介绍

     在发现有些commit出错时,可使用Hudi提供的rollback回滚至指定的commit,这样可防止出现错误的结果,并且当一次commit失败时,也会进行rollback操作,保证一次commit的原子性。

2. 分析

     rollback(回滚)的入口在 HoodieWriteClient#rollback,其依赖 HoodieWriteClient#rollbackInternal方法完成实际的回滚,其核心代码如下

 

protected void rollbackInternal(String commitToRollback) {

// 生成新的rollback时间

final String startRollbackTime = HoodieActiveTimeline.createNewInstantTime();

try {

HoodieTable<T> table = HoodieTable.getHoodieTable(

createMetaClient(true), config, jsc);

// 找出第一个与rollback commit相等的instant

Option<HoodieInstant> rollbackInstantOpt =

Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstants()

.filter(instant -> HoodieActiveTimeline.EQUAL.test(instant.getTimestamp(), commitToRollback))

.findFirst());

// 存在

if (rollbackInstantOpt.isPresent()) {

// 进行回滚

List<HoodieRollbackStat> stats = doRollbackAndGetStats(rollbackInstantOpt.get());

// 结束回滚

finishRollback(context, stats, Collections.singletonList(commitToRollback), startRollbackTime);

}

} catch (IOException e) {

throw new HoodieRollbackException("Failed to rollback " + config.getBasePath() + " commits " + commitToRollback,

e);

}

}

 

        首先过滤出commit/delta_commit中是否存在待回滚instant的时间,如果存在,则进行回滚,回滚的核心方法为 doRollbackAndGetStats,该方法在前一篇讲解savepoint时已经分析过,该方法会调用 HoodieTable#rollback完成实际回滚动作,下面着重分析 HoodieTable#rollback方法,对于MOR和COW不同类型有不同实现,下面一一进行分析。

 

2.1 HoodieCopyOnWriteTable#rollback

对于COW类型而言, rollback核心代码如下

public List<HoodieRollbackStat> rollback(JavaSparkContext jsc, HoodieInstant instant, boolean deleteInstants)

throws IOException {

long startTime = System.currentTimeMillis();

List<HoodieRollbackStat> stats = new ArrayList<>();

HoodieActiveTimeline activeTimeline = this.getActiveTimeline();



if (instant.isCompleted()) { // instant状态为completed

// 转变至inflight状态

instant = activeTimeline.revertToInflight(instant);

}



if (!instant.isRequested()) { // 不为requested状态

String commit = instant.getTimestamp();

// 生成回滚的请求

List<RollbackRequest> rollbackRequests = generateRollbackRequests(instant);

// 进行回滚

stats = new RollbackExecutor(metaClient, config).performRollback(jsc, instant, rollbackRequests);

}

// 删除inflight和requested状态的instant

deleteInflightAndRequestedInstant(deleteInstants, activeTimeline, instant);

return stats;

}

 

可以看到,进行回滚总体分为四步:

1. 对于处理completed状态的instant,首先会将其转变至inflight状态,而对于不处于requested状态的instant(compaction会存在requested状态);

2. 生成回滚请求;

3. 进行回滚;

4. 删除instant。

 

2.1.1 转变instant状态

对于处于completed状态的instant,将其转变至 inflight状态,其核心代码如下

public HoodieInstant revertToInflight(HoodieInstant instant) {

// 获取inflight状态的instant

HoodieInstant inflight = HoodieTimeline.getInflightInstant(instant, metaClient.getTableType());

// 转变至inflight,即文件名会变为.inflight

revertCompleteToInflight(instant, inflight);

return inflight;

}

 

对于状态转变体现在文件名后缀的变化,即会变为 .inflght状态。

 

2.1.2 生成回滚请求

回滚请求由 generateRollbackRequests方法生成,其核心代码如下

private List<RollbackRequest> generateRollbackRequests(HoodieInstant instantToRollback)

throws IOException {

// 获取所有的分区路径,对每个分区路径生成DELETE_DATA_AND_LOG_FILES类型的RollbackRequest

return FSUtils.getAllPartitionPaths(this.metaClient.getFs(), this.getMetaClient().getBasePath(),

config.shouldAssumeDatePartitioning()).stream().map(partitionPath -> RollbackRequest.createRollbackRequestWithDeleteDataAndLogFilesAction(partitionPath, instantToRollback))

.collect(Collectors.toList());

}

 

会根据不同的分区路径生成不同的RollbackRequest,该方法会生成会生成DELETEDATAANDLOGFILES类型,指定分区路径的RollbackRequest。

 

2.1.3 进行回滚

通过 RollbackExecutor#performRollback进行回滚,其核心代码如下

对于DELETEDATAFILES_ONLY类型的rollback,会调用 deleteCleanedFiles来删除数据文件,其核心代码如下

private Map<FileStatus, Boolean> deleteCleanedFiles(HoodieTableMetaClient metaClient, HoodieWriteConfig config,

Map<FileStatus, Boolean> results, String commit, String partitionPath) throws IOException {

FileSystem fs = metaClient.getFs();

PathFilter filter = (path) -> {

if (path.toString().contains(".parquet")) { // 数据文件

String fileCommitTime = FSUtils.getCommitTime(path.getName());

// 与rollback时间相等

return commit.equals(fileCommitTime);

}

return false;

};

// 过滤出与rollback时间相等的所有parquet文件

FileStatus[] toBeDeleted = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter);

for (FileStatus file : toBeDeleted) { // 逐一删除

boolean success = fs.delete(file.getPath(), false);

results.put(file, success);

}

return results;

}

 

首先会过滤指定分区下所有与rollback时间相等的parquet文件,然后逐一删除。

对于DELETEDATAANDLOGFILES类型的rollback,会调用同名的 deleteCleanedFiles来删除文件,其核心代码如下

private Map<FileStatus, Boolean> deleteCleanedFiles(HoodieTableMetaClient metaClient, HoodieWriteConfig config,

Map<FileStatus, Boolean> results, String partitionPath, PathFilter filter) throws IOException {

FileSystem fs = metaClient.getFs();

// 过滤出与rollback时间相等的所有parquet和log文件

FileStatus[] toBeDeleted = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter);

for (FileStatus file : toBeDeleted) { // 逐一删除

boolean success = fs.delete(file.getPath(), false);

results.put(file, success);

}

return results;

}

 

首先会过滤指定分区下所有与rollback时间相等的parquet/log文件,然后逐一删除。

对于APPENDROLLBACKBLOCK类型的rollback,会生成日志文件控制块并写入指定的文件中,在读取时,将不会读取该控制块的前一个块。

 

2.1.4 删除instant

在完成回滚后,还需要调用 deleteInflightAndRequestedInstant来删除instant,其核心代码如下

protected void deleteInflightAndRequestedInstant(boolean deleteInstant, HoodieActiveTimeline activeTimeline,

HoodieInstant instantToBeDeleted) {

// 删除marker下的目录

deleteMarkerDir(instantToBeDeleted.getTimestamp());



if (deleteInstant) { // 删除instant

// 删除处于pending状态的instant

activeTimeline.deletePending(instantToBeDeleted);

if (instantToBeDeleted.isInflight() && !metaClient.getTimelineLayoutVersion().isNullVersion()) {

// 删除处于requested状态的instant

instantToBeDeleted = new HoodieInstant(State.REQUESTED, instantToBeDeleted.getAction(),

instantToBeDeleted.getTimestamp());

activeTimeline.deletePending(instantToBeDeleted);

}

}

}

 

删除instant主要是删除处于inflight和requested状态的在元数据目录下的文件。

 

2.2 HoodieMergeOnReadTable#rollback

对于MOR而言, rollback核心代码如下

public List<HoodieRollbackStat> rollback(JavaSparkContext jsc, HoodieInstant instant,

boolean deleteInstants) throws IOException {

long startTime = System.currentTimeMillis();

// rollback时间

String commit = instant.getTimestamp();



if (instant.isCompleted()) { // instant状态为completed

// // 转变至inflight状态

instant = this.getActiveTimeline().revertToInflight(instant);

}



List<HoodieRollbackStat> allRollbackStats = new ArrayList<>();



// 不为requested状态

if (!instant.isRequested()) {

// 生成rollback请求

List<RollbackRequest> rollbackRequests = generateRollbackRequests(jsc, instant);

// 进行回滚

allRollbackStats = new RollbackExecutor(metaClient, config).performRollback(jsc, instant, rollbackRequests);

}



// 删除inflight和requested状态的instant

deleteInflightAndRequestedInstant(deleteInstants, this.getActiveTimeline(), instant);



return allRollbackStats;

}

可以看到其流程与COW相同,不再赘述。

3. 总结

      对于rollback而言,其主要分为四步:转变instant状态;2. 生成回滚请求;3. 进行回滚;4. 删除instant。而回滚时会分为三种情况,对于 DELETE_DATA_FILES_ONLYDELETE_DATA_AND_LOG_FILES类型的rollback,会直接删除对应commit的数据文件和日志文件,而对于 APPEND_ROLLBACK_BLOCK类型,则会写入控制块至文件中,在读取时不读取其前一个块。

相关标签: hudi Hudi