Cannot process data using Spark Continuous Streaming

user_14354117 0
Viewed 1453

I'm developing a real-time streaming application that polls data from a Kafka broker, and I am adapting the code that previously worked using Spark Structured Streaming by default (with micro-batching). However, I don't know how to obtain a similar behavior using continuous streaming instead of micro-batching streaming. This is the piece of code that works:

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

This is what I tried so far with continuous streaming:

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

And the app pops out the following error:

Continuous execution does not support task retry at org.apache.spark.sql.execution.streaming.continuous.ContinuousDataSourceRDD.compute(ContinuousDataSourceRDD.scala:76)

I'm using Spark (pyspark) 3.0.1 w/Scala 2.12, Kafka 2.6.0

and I'm adding jar org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1 when I submit the app.

spark-structured-streaming pyspark apache-kafka apache-spark
2 Answers

The continuous processing mode in Spark Structured Streaming is only available for some query types.

According to the documentation on Continuous Processing the following queries are supported and it seems that your query is not supported:

As of Spark 2.4, only the following type of queries are supported in the continuous processing mode.

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.
user_122089100

If you see this exception, it's not root cause of job failure. You should check other task failure reason. You can try spark.task.maxFailures=1, this will depress ContinuousTaskRetryException, it's easier to identify real root cause.

Use Continuous Processing mode only when all below conditions meet:

  1. Want to keep end-to-end latency strictly below 100ms
  2. At least once processing is acceptable, which means message could be processed multiple times
  3. Could tolerate manually resubmit job when any task or executor failed since cp mode does not support task retry

My suggestions is not use cp mode unless you really need such low latency as cp mode is still experimental. More details please refer https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing

user_29243730
Related