[ https://issues.apache.org/jira/browse/SPARK-40821?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Jungtaek Lim updated SPARK-40821: --------------------------------- Affects Version/s: 3.4.0 (was: 3.3.0) > Fix late record filtering to support chaining of steteful operators > ------------------------------------------------------------------- > > Key: SPARK-40821 > URL: https://issues.apache.org/jira/browse/SPARK-40821 > Project: Spark > Issue Type: Improvement > Components: Structured Streaming > Affects Versions: 3.4.0 > Reporter: Alex Balikov > Assignee: Alex Balikov > Priority: Major > Fix For: 3.4.0 > > > Currently chaining of stateful operators is Spark Structured Streaming is not > supported for various reasons and is blocked by the unsupported operations > check (spark.sql.streaming.unsupportedOperationCheck flag). We propose to fix > this as chaining of stateful operators is a common streaming scenario - e.g. > stream-stream join -> windowed aggregation > window aggregation -> window aggregation > etc > What is broken: > # every stateful operator performs late record filtering against the global > watermark. When chaining stateful operators (e.g. window aggregations) the > output produced by the first stateful operator is effectively late against > the watermark and thus filtered out by the next operator late record > filtering (technically the next operator should not do late record filtering > but it can be changed to assert for correctness detection, etc) > # when chaining window aggregations, the first window aggregating operator > produces records with schema \{ window: { start: Timestamp, end: Timestamp }, > agg: Long } - there is not explicit event time in the schema to be used by > the next stateful operator (the correct event time should be window.end - 1 ) > # stream-stream time-interval join can produce late records by semantics, > e.g. if the join condition is: > left.eventTime BETWEEN right.eventTime + INTERVAL 1 HOUR right.eventTime - > INTERVAL 1 HOUR > the produced records can be delayed by 1 hr relative to the > watermark. > Proposed fixes: > 1. 1 can be fixed by performing late record filtering against the previous > microbatch watermark instead of the current microbatch watermark. > 2. 2 can be fixed by allowing the window and session_window functions to work > on the window column directly and compute the correct event time > transparently to the user. Also, introduce window_time SQL function to > compute correct event time from the window column. > 3. 3 can be fixed by adding support for per-operator watermarks instead of a > single global watermark. In the example of stream-stream time interval join > followed by a stateful operator, the join operator will 'delay' the > downstream operator watermarks by a correct value to handle the delayed > records. Only stream-stream time-interval joins will be delaying the > watermark, any other operators will not delay downstream watermarks. > > -- This message was sent by Atlassian Jira (v8.20.10#820010) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org