scala - 结构化流 - 无法使用 FileContext API 管理 AWS S3 上的元数据日志文件

Question

我在 Spark(v2.2.0) 中有一个 StreamingQuery，即，

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "test")
  .load()

val query = df
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("parquet")
  .option("checkpointLocation", "s3n://bucket/checkpoint/test")
  .option("path", "s3n://bucket/test")
  .start()

当我运行 query 时，数据会保存在 AWS S3 上，并且会在 s3n://bucket/checkpoint/test 中创建检查点。但是，我还在日志中收到以下警告:

WARN [o.a.s.s.e.streaming.OffsetSeqLog] Could not use FileContext API for managing metadata log files at path s3n://bucket/checpoint/test/offsets. Using FileSystem API instead for managing log files. The log may be inconsistent under failures.

我无法理解为什么会出现此警告。另外，如果出现任何故障，我的检查点是否会不一致？

谁能帮我解决一下？

Answer 1

查看源码，这个错误来自HDFSMetadataLog类(class)。代码中的注释指出:

Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing files in a directory always shows the latest files.

所以问题是由于使用 AWS S3 而导致的，它会强制您使用 FileSystemManager API。检查该类的评论，我们看到，

Implementation of FileManager using older FileSystem API. Note that this implementation cannot provide atomic renaming of paths, hence can lead to consistency issues. This should be used only as a backup option, when FileContextManager cannot be used.

因此，当多个编写器想要同时进行重命名操作时，可能会出现一些问题。有相关工单here ，但是，由于无法在 Spark 中修复该问题，因此已将其关闭。

如果需要在 S3 上设置检查点，需要考虑的一些事项:

1. 为避免警告和潜在问题，检查点到 HDFS，然后复制结果
2. 到 S3 的检查点，但检查点之间有很长的间隔。