hadoop - Flink - AWS EMR 中的 AWSS3IOException 由带有 S3A 的 BucketingSink 引起

Question

我有一个在 AWS EMR 中运行的具有高并行度 (400) 的 Flink 应用程序。它使用 BucketingSink(使用 RocksDb 后端进行检查点)获取 Kafka 并汇入 S3。目的地使用“s3a://”前缀定义。 Flink 作业是一个连续运行的流式应用程序。在任何给定时间，所有工作人员加起来可能会生成/写入 400 个文件(由于 400 并行度)。几天后，其中一名 worker 将失败，但出现异常:

org.apache.hadoop.fs.s3a.AWSS3IOException: copyFile(bucket/2018-09-01/05/_file-10-1.gz.in-progress, bucket/2018-09-01/05/_file-10-1.gz.pending): com.amazonaws.services.s3.model.AmazonS3Exception: We encountered an internal error. Pelase try again. (Service: Amazon S3; Status Code: 200 InternalError; Request ID: xxxxxxxxxx; S3 Extended Request ID: yyyyyyyyyyyyyyy
at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java: 178)
at org.apache.hadoop.fs.s3a.S3AFileSystem.copyFile(S3AFileSystem.java: 1803)
at org.apache.hadoop.fs.s3a.S3AFileSystem.innerRename(S3AFileSystem.java:776)
at org.apache.hadoop.fs.s3a.S3AFileSystem.rename(S3AFileSystem.java:662)
at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.closeCurrentPartFile(BucketingSink.java:575)
at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.openNewPartFile(BucketingSink.java:514)
at org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.invoke(BucketingSink.java:446)

这似乎是在 BucketingSink 创建新的零件文件时随机发生的。奇怪的是，这是随机发生的，当它发生时，它发生在并行 flink worker 中的 1 个(不是全部)。此外，当发生这种情况时，Flink 作业会转换为 FAILING 状态，但 Flink 作业不会重新启动并从上一个成功的检查点恢复/恢复。这是什么原因造成的，应该如何解决？此外，如何将作业配置为从上一个成功的检查点重新启动/恢复，而不是保持在失败状态？

Answer 1

我认为这是 bucketing sink 和 S3 的已知行为，建议的解决方案是使用 Shiny 的新 StreamingFileSink在 Flink 1.7.0 中。

基本上，bucketing sink 期望写入和重命名像在真实文件系统中一样立即发生，但这对于像 S3 这样的对象存储来说不是一个好的假设，因此 bucketing sink 最终会出现导致间歇性中断的竞争条件问题。这是一张描述问题的 JIRA 票证，相关的票证进一步充实了它。 JIRA FLINK-9752