Versions Compared

Old Version 6

changes.mady.by.user Almog Gavra

Saved on

compared with

New Version Current

changes.mady.by.user Almog Gavra

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Code Block
public final class OffsetOrderedWindowsBatchWindows extends Windows<TimeWindow> {
		
		/**
		 * Return a window definition with the given window size.
		 * <p> 
		 * This represents stream time"batching" window semantics, which are fixed-size,
		 * gap-less, non-overlapping windows. StaticBatched windows use the current
		 * stream time when determining which window an event fits into instead
		 * of the event time (as the other window types use). This allows users
         * to batch together records of the same key over a window of time.
		 * <p>
 		 * The window boundaries are inclusive of the start point and exclusive,
		 * of the end point. This means that a BatchWindows of size 1000 would have
		 * windows from [0,1000),[1000,2000),etc...
         * <p>
         * Note that offset orderedbatch windows will always accept late arriving data, and
		 * should only be used in situations where the ordering of data is not
		 * necessarily important to the semantics of your application.
		 */
		public static OffsetOrderedWindowsBatchWindows ofSize(final Duration size) { ... }
    
}

Here is an example of how events would map to windows:

Image RemovedImage Added


The example in the motivation section then becomes:

Code Block
stream
	.windowBy(OffsetOrderedWindowsBatchWindows.ofSize(Duration.ofSeconds(10))
	.aggregate((k, v, kvList) -> kvList.add((k, v)))
	.suppress(Suppressed.untilWindowCloses(...))

...

Code Block
public abstract class Windows<W extends Window> {
  ...
  public abstract Map<Long, W> windowsFor(
    final long timestamp, 
    final long observedStreamTime
  ) {
    return windowsFor(timestamp);
  }

  // as part of this KIP we will deprecate this method, which is technically public
  // although users are not expected to implement this (it lives in a public package)
  @Deprecated
  public abstract Map<Long, W> windowsFor(final long timestamp);
}

The implementation of OffsetOrderedWindows BatchWindows will then use the observedStreamTime when computing which window the current event should fall into:

Code Block
@Override
public Map<Long, TimeWindow> windowsFor(
  final long timestamp, 
  final long observedStreamTime
) {
	long windowStart = (observedStreamTime / sizeMs) * sizeMs;
  return Map.of(windowStart, new TimeWindow(windowStart, windowStart + sizeMs));
}

The gracePeriodMs for OffsetOrderedWindows BatchWindows is always zero, which means there will only ever be one open window per key and events will always fall within this window. Here is the documentation for the remaining public members of the class:

Code Block
/**
 * A fixed-size, stream-time based window specification used for aggregations.
 * <p>
 * The semantics of offset orderedbatched aggregations are: Every size() milliseconds, compute the aggregate total for the last
 * size() milliseconds. This is equivalent in semantics to TimeWindows with an advance of 0.
 * <p>
 * This class differs from TimeWindows in that the windows for a given event is computed based on the current observed
 * stream time as opposed to the timestamp of the given record. This effectively makes the windows ordered with respect
 * to the offset of the record instead of the timestamp and allows for batching together records within a window.
 */
public final class OffsetOrderedWindowsBatchWindows extends Windows<TimeWindow> {
  
  /**
   * Returns the window size, measured in milliseconds.
   */
  public long size() { }

  /**
   * Returns 0, as offset ordered windows have no concept of "late arriving" records
   * because stream time is always monotonically increasing
   */
   public long gracePeriodMs() { return 0L; }

}

...

  • Instead of introducing a new window type, we could add a property like Strictness  on the existing TimeWindows  type. This would end up being more confusing than the proposed API since grace period do not apply to the type of windows suggested here.
  • Names we discussed but didn't like: GlobalWindow, StreamTimeWindow, ProcessTimeWindow, OffsetOrderedWindow.
  • Implementing a more general purpose batching mechanism that can batch based on number of records, wall clock or any other batching strategy. This is useful, but out of scope for this proposal as it introduces a different level of non-determinism that is not covered by the existing grace/windowing semantics. The way batching/suppression interrelate are also