将 Spark 结构化流与 Trigger.Once 结合使用

Question

有一个全天更新的 CSV 文件数据湖。我正在尝试使用 Trigger.Once 功能 outlined in this blog post 创建 Spark 结构化流式处理作业，以定期将已写入 Parquet 数据湖中 CSV 数据湖的新数据写入。

这是我所拥有的：

val df = spark
  .readStream
  .schema(s)
  .csv("s3a://csv-data-lake-files")

以下命令将所有数据写入 Parquet 湖，但在所有数据写入后并没有停止（我不得不手动取消作业）。

processedDf
  .writeStream
  .trigger(Trigger.Once)
  .format("parquet")
  .option("checkpointLocation", "s3-path-to-checkpoint")
  .start("s3-path-to-parquet-lake")

以下作业也有效，但在写入所有数据后也没有停止（我不得不手动取消作业）：

val query = processedDf
  .writeStream
  .trigger(Trigger.Once)
  .format("parquet")
  .option("checkpointLocation", "s3-path-to-checkpoint")
  .start("s3-path-to-parquet-lake")

query.awaitTermination()

以下命令在写入任何数据之前停止查询。

val query = processedDf
  .writeStream
  .trigger(Trigger.Once)
  .format("parquet")
  .option("checkpointLocation", "s3-path-to-checkpoint")
  .start("s3-path-to-parquet-lake")

query.stop()

如何将writeStream 查询配置为等到所有增量数据都写入 Parquet 文件然后停止？

Answer 1

我得到了结构化流 + 触发器。曾经在 Parquet 数据湖上正常工作。

我认为它不适用于 CSV 数据湖，因为 CSV 数据湖在嵌套目录中有大量小文件。 Spark 不喜欢处理小型 CSV 文件（我认为它需要将它们全部打开才能读取标题）并且非常讨厌需要全局 S3 目录。

所以我认为 Spark Structured Streaming + Trigger.Once 代码很好 - 他们只需要改进 CSV 阅读器技术。

Answer 2

结构化流的主要目的是连续处理数据，而无需在新数据到达时启动/停止流。阅读this了解更多详情。

从 Spark 2.0.0 开始 StreamingQuery 有方法 processAllAvailable 等待所有源数据被处理并提交到接收器。请注意 scala 文档states to use this method for testing purpose only.

因此代码应该是这样的（如果你还想要的话）：

query.processAllAvailable
query.stop