无法使用 Spark 连续流处理数据答案

【问题标题】：Cannot process data using Spark Continuous Streaming无法使用 Spark 连续流处理数据
【发布时间】：2021-01-14 23:12:53
【问题描述】：

我正在开发一个实时流应用程序，该应用程序从 Kafka 代理轮询数据，并且我正在调整以前默认使用 Spark 结构化流处理的代码（使用微批处理）。但是，我不知道如何使用连续流而不是微批处理流来获得类似的行为。这是一段有效的代码：

query = df.writeStream \
        .foreachBatch(foreach_batch_func) \
        .start()

这是我迄今为止尝试的连续流式传输：

query = df \
        .writeStream \
        .foreach(example_func) \
        .trigger(continuous = '1 second') \
        .start()

应用弹出如下错误：

连续执行不支持org.apache.spark.sql.execution.streaming.continuous.ContinuousDataSourceRDD.compute(ContinuousDataSourceRDD.scala:76)处的任务重试

我正在使用带有 Scala 2.12、Kafka 2.6.0 的 Spark (pyspark) 3.0.1

当我提交应用程序时，我正在添加 jar org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1。

【问题讨论】：

标签： apache-spark pyspark apache-kafka spark-structured-streaming

【解决方案1】：

如果您看到此异常，这不是作业失败的根本原因。您应该检查其他任务失败的原因。你可以试试 spark.task.maxFailures=1，这会抑制 ContinuousTaskRetryException，更容易找出真正的根本原因。

仅当满足以下所有条件时才使用连续处理模式：

希望将端到端延迟严格控制在 100 毫秒以下
至少一次处理是可以接受的，这意味着消息可以多次处理
由于 cp 模式不支持任务重试，当任何任务或执行程序失败时可以手动重新提交作业

我的建议是不要使用 cp 模式，除非你真的需要如此低的延迟，因为 cp 模式仍处于试验阶段。更多详情请参考https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing

【讨论】：

【解决方案2】：

Spark Structured Streaming 中的连续处理模式仅适用于部分查询类型。

根据Continuous Processing 上的文档，支持以下查询，但您的查询似乎不受支持：

从 Spark 2.4 开始，连续处理模式仅支持以下类型的查询。

Operations: Only map-like Dataset/DataFrame operations are supported in continuous mode, that is, only projections (select, map, flatMap, mapPartitions, etc.) and selections (where, filter, etc.).
   All SQL functions are supported except aggregation functions (since aggregations are not yet supported), current_timestamp() and current_date() (deterministic computations using time is challenging).
Sources:
   Kafka source: All options are supported.
   Rate source: Good for testing. Only options that are supported in the continuous mode are numPartitions and rowsPerSecond.
Sinks:
   Kafka sink: All options are supported.
   Memory sink: Good for debugging.
   Console sink: Good for debugging. All options are supported. Note that the console will print every checkpoint interval that you have specified in the continuous trigger.

【讨论】：