如何使作业幂等于多次运行在S3中生成相同的结果文件

Posted 2021-05-07

我使用spark df write将镶木地板文件保存到S3。场景：我运行Job1并将xyz.parquet保存到S3，我的Job2拿起xyz.parquet并将数据加载到DB。现在，我希望下次运行Job1时，它应生成xyz.parquet并将数据加载到DB。我的问题是，我无法覆盖镶木地板文件，我不想在保存新的xyz.parquet之前删除旧的xyz.parquet文件因为可能有新的xyz.parquet可能无法写入的机会不想失去我的旧xyz.parquet。我希望我的Job1是幂等的，这样我的Job2和其他工作就不需要改变了。一种方法是在时间戳附加的文件名的命名约定中。有没有更好的解决方案。

答案

一般工作流程从源目录dir重命名为dest。 S3的特殊之处在于，当您写入文件时，在调用文件close（）之前，没有任何内容（可见）保存。也就是说：如果任何程序在覆盖文件中途失败，那么旧文件仍然存在。也许你可以将这个功能用于像这样的工作流程

1. 使用输出s3：//bucket/tmp/result.parquet运行查询
2. 发出s3 HTTP请求PUT s3://bucket/final/result.parquet x-copy-source:s3://bucket/tmp/result.parquet

S3的Hadoop FS客户端都没有直接公开该复制命令，虽然重命名是这样做，但强制你先删除最后一个，所以那里有一个很小的漏洞