You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "padavan (Jira)" <ji...@apache.org> on 2023/04/01 07:54:00 UTC
[jira] [Updated] (SPARK-43001) Spark last window dont flush in append mode
[ https://issues.apache.org/jira/browse/SPARK-43001?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
padavan updated SPARK-43001:
----------------------------
Description:
The problem is very simple, when you use *TUMBLING* *window* with {*}append mode{*}, then the window is closed only when the next message arrives ({_}+watermark logic{_}).
In the current implementation, if you *stop* *incoming* streaming data, the *last* window will *NEVER close* and we LOSE the last window data.
Business situation:
Worked correctly and new messages stop incoming and next message come in 5 hours later and the client will get the message after 5 hours instead of the 10 seconds delay of window.
!https://user-images.githubusercontent.com/61819835/226478055-dc4a123c-4397-4eb0-b6ed-1e185b6fab76.png|width=971,height=404!
The current implementation needs to be improved. Include in spark internal mechanisms to close windows automatically.
*What we propose:*
Add third parameter
{{{}DataFrame.{}}}{{{}withWatermark{}}}({_}eventTime{_}, {_}delayThreshold, *maxDelayClose*{_}). And then trigger will execute
{code:java}
if(now-window.upper_bound > maxDelayClose){
window.close().flush();
}
{code}
I assume it can be done in a day. It wasn't expected for us that our customers couldn't get the notifications. (the company is in the medical field).
!image-2023-04-01-10-38-43-326.png|width=159,height=101!
simple code for problem:
{code:java}
kafka_stream_df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", KAFKA_BROKER) \
.option("subscribe", KAFKA_TOPIC) \
.option("includeHeaders", "true") \
.load()
sel = (kafka_stream_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.select(from_json(col("value").cast("string"), json_schema).alias("data"))
.select("data.*")
.withWatermark("dt", "1 seconds")
.groupBy(window("dt", "10 seconds"))
.agg(sum("price"))
)
console = sel \
.writeStream \
.trigger(processingTime='10 seconds') \
.format("console") \
.outputMode("append")\
.start()
{code}
was:
The problem is very simple, when you use *TUMBLING* *window* with {*}append mode{*}, then the window is closed only when the next message arrives ({_}+watermark logic{_}).
In the current implementation, if you *stop* *incoming* streaming data, the *last* window will *NEVER close* and we LOSE the last window data.
Business situation:
Worked correctly and new messages stop incoming and next message come in 5 hours later and the client will get the message after 5 hours instead of the 10 seconds delay of window.
!https://user-images.githubusercontent.com/61819835/226478055-dc4a123c-4397-4eb0-b6ed-1e185b6fab76.png!
The current implementation needs to be improved. Include in spark internal mechanisms to close windows automatically.
*What we propose:*
Add third parameter
{{{}DataFrame.{}}}{{{}withWatermark{}}}({_}eventTime{_}, {_}delayThreshold, *maxDelayClose*{_}). And then trigger will execute
{code:java}
if(now-window.upper_bound > maxDelayClose){
window.close().flush();
}
{code}
I assume it can be done in a day. It wasn't expected for us that our customers couldn't get the notifications. (the company is in the medical field).
!image-2023-04-01-10-38-43-326.png|width=159,height=101!
simple code for problem:
{code:java}
kafka_stream_df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", KAFKA_BROKER) \
.option("subscribe", KAFKA_TOPIC) \
.option("includeHeaders", "true") \
.load()
sel = (kafka_stream_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.select(from_json(col("value").cast("string"), json_schema).alias("data"))
.select("data.*")
.withWatermark("dt", "1 seconds")
.groupBy(window("dt", "10 seconds"))
.agg(sum("price"))
)
console = sel \
.writeStream \
.trigger(processingTime='10 seconds') \
.format("console") \
.outputMode("append")\
.start()
{code}
> Spark last window dont flush in append mode
> -------------------------------------------
>
> Key: SPARK-43001
> URL: https://issues.apache.org/jira/browse/SPARK-43001
> Project: Spark
> Issue Type: Bug
> Components: PySpark, Spark Core
> Affects Versions: 3.3.2
> Reporter: padavan
> Priority: Critical
> Original Estimate: 24h
> Remaining Estimate: 24h
>
> The problem is very simple, when you use *TUMBLING* *window* with {*}append mode{*}, then the window is closed only when the next message arrives ({_}+watermark logic{_}).
> In the current implementation, if you *stop* *incoming* streaming data, the *last* window will *NEVER close* and we LOSE the last window data.
>
> Business situation:
> Worked correctly and new messages stop incoming and next message come in 5 hours later and the client will get the message after 5 hours instead of the 10 seconds delay of window.
> !https://user-images.githubusercontent.com/61819835/226478055-dc4a123c-4397-4eb0-b6ed-1e185b6fab76.png|width=971,height=404!
> The current implementation needs to be improved. Include in spark internal mechanisms to close windows automatically.
>
> *What we propose:*
> Add third parameter
> {{{}DataFrame.{}}}{{{}withWatermark{}}}({_}eventTime{_}, {_}delayThreshold, *maxDelayClose*{_}). And then trigger will execute
>
> {code:java}
> if(now-window.upper_bound > maxDelayClose){
> window.close().flush();
> }
> {code}
>
> I assume it can be done in a day. It wasn't expected for us that our customers couldn't get the notifications. (the company is in the medical field).
> !image-2023-04-01-10-38-43-326.png|width=159,height=101!
>
>
> simple code for problem:
>
> {code:java}
> kafka_stream_df = spark \
> .readStream \
> .format("kafka") \
> .option("kafka.bootstrap.servers", KAFKA_BROKER) \
> .option("subscribe", KAFKA_TOPIC) \
> .option("includeHeaders", "true") \
> .load()
> sel = (kafka_stream_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
> .select(from_json(col("value").cast("string"), json_schema).alias("data"))
> .select("data.*")
> .withWatermark("dt", "1 seconds")
> .groupBy(window("dt", "10 seconds"))
> .agg(sum("price"))
> )
>
> console = sel \
> .writeStream \
> .trigger(processingTime='10 seconds') \
> .format("console") \
> .outputMode("append")\
> .start()
> {code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org