1

我对获取 S3 parquet 数据并将 parquet 数据写入 S3 的 Spark Streaming 应用程序 (Spark v2.3.2) 感兴趣。应用程序的数据帧流使用groupByKey()并flatMapGroupsWithState()使用GroupState.

是否可以将其配置为使用s3检查点位置？例如：

val stream = myDataset.writeStream
    .format("parquet")
    .option("path", s3DataDestination)
    .option("checkpointLocation", s3CheckpointPath)
    .option("truncate", false)
    .option(Trigger.Once)
    .outputMode(OutputMode.Append)
stream.start().awaitTermination()

我确认以上是能够成功将数据写入s3DataDestination.

但是，写入 s3 检查点位置时会引发异常：

java.lang.IllegalStateException: Error committing version 1 into HDFSStateStore[id=(op=0, part=9), dir=s3://<my_s3_location>
    at org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider$HDFSBackedStateStore.commit(...)
...
Caused by: java.io.IOException: Failed to rename s3://.../checkpoint/state/0/9/temp... to s3://.../checkpoint/state/0/9/1.delta

这需要自定义实现 S3StateStoreProvider吗？或者，检查点位置是否需要写入 HDFS？

4

1 回答 1

1

问题是写入和读取的并发频率太高。AWS S3 不提供此类功能。

解决方案 ：

  • 我们必须切换到本地安装的永久磁盘以进行 Spark 检查点
  • S3Guard：这将使 S3 的读写更加一致（注意：这是实验性的，我个人从未见过它实际使用过）
  • 使用 HDFS
于 2020-06-18T03:19:14.347 回答