Uploaded image for project: 'Spark'
  1. Spark
  2. SPARK-40821

Introduce window_time function to extract event time from the window column



    • Improvement
    • Status: Resolved
    • Major
    • Resolution: Fixed
    • 3.4.0
    • 3.4.0
    • Structured Streaming
    • None


      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


      What is broken:

      1. 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)
      2. 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 )

      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 ticket handles no. 2 of the proposal. Others will be handled in separate ticket.





            alex-balikov Alex Balikov
            alex-balikov Alex Balikov
            Jungtaek Lim Jungtaek Lim
            0 Vote for this issue
            3 Start watching this issue