[
https://issues.apache.org/jira/browse/SPARK-42102?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon resolved SPARK-42102.
----------------------------------
Resolution: Invalid
Resolving as Invalid — this is a usage/how-to question rather than a specific
Spark defect or actionable change. Usage questions are best directed to
[email protected] (https://spark.apache.org/community.html) or Stack
Overflow (tag apache-spark). Findings from triage: The ticket is explicitly
typed "Question" with 0 comments, asking "how to use checkpoints with
foreachBatch stateful transformation" — a pure how-to inquiry containing no bug
report, reproducer, or actionable change request. Both behaviors are documented
and verified in the current source:
docs/streaming/apis-on-dataframes-and-datasets.md documents the
checkpointLocation option (numerous examples, e.g. lines 2245/2288/2331/2371)
and the foreachBatch sink section (~2391-2483), including at-least-once vs
exactly-once via batchId dedup, continuous-mode limitations, and explicit
stateful-query per
Please reopen with a concrete reproducer or a specific proposed change if this
is actually a bug or an actionable improvement.
> Using checkpoints in Spark Structured Streaming with the foreachBatch sink
> --------------------------------------------------------------------------
>
> Key: SPARK-42102
> URL: https://issues.apache.org/jira/browse/SPARK-42102
> Project: Spark
> Issue Type: Question
> Components: PySpark, Structured Streaming
> Affects Versions: 3.3.1
> Reporter: Kai-Michael Roesner
> Priority: Major
>
> I want to build a fault-tolerant, recoverable Spark job (using Structured
> Streaming in PySpark) that reads a data stream from Kafka and uses the
> [{{foreachBatch}}|https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreachbatch]
> sink to implement a stateful transformation before writing the resulting
> data to the actual sink.
> The basic structure of my Spark job is like this:
> {code}
> counter = 0
> def batch_handler(df, batch_id):
> global counter
> counter += 1
> df.withColumn('counter', lit(counter)).show(truncate=30)
> spark = (SparkSession.builder
> .appName('test.stateful.checkpoint')
> .config('spark.jars.packages', f'{KAFKA_SQL},{KAFKA_CLNT}')
> .getOrCreate())
> source = (spark.readStream
> .format('kafka')
> .options(**KAFKA_OPTIONS)
> .option('subscribe', 'topic-spark-stateful')
> .option('startingOffsets', 'earliest')
> .option('includeHeaders', 'true')
> .load())
> (source
> .selectExpr('CAST(value AS STRING) AS data', 'CAST(timestamp AS STRING) AS
> time')
> .writeStream
> .option('checkpointLocation', './checkpoints/stateful')
> .foreachBatch(batch_handler)
> .start()
> .awaitTermination())
> {code}
> where the simplified {{batch_handler}} function is a stand-in for the
> stateful transformation + writer to the actual data sink. Also for simplicity
> I am using a local folder as checkpoint location.
> This works fine as far as checkpointing of Kafka offsets is concerned. But
> how can I include the state of my custom batch handler ({{counter}} in my
> simplified example) in the checkpoints such that the job can pick up where it
> left after a crash?
> The [Spark Structured Streaming
> Guide|https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing]
> doesn't say anything on the topic. With the
> [{{foreach}}|(https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreach]
> sink I can pass a custom row handler object but this seems to support only
> {{open}}, {{process}}, and {{close}} methods.
> Would it make sense to create a "Request" or even "Feature" ticket to enhance
> this with methods for restoring state from a checkpoint and exporting state
> to support checkpointing?
> PS: I have posted this on [SOF|https://stackoverflow.com/questions/74864425],
> too. If anyone cares to answer or comment I'd be happy to upvote their post.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
