public final class JoinWindows extends Windows<Window>
A JoinWindows instance defines a maximum time difference for a join over two streams on the same key.
In SQL-style you would express this join as
SELECT * FROM stream1, stream2
WHERE
stream1.key = stream2.key
AND
stream1.ts - before <= stream2.ts AND stream2.ts <= stream1.ts + after
Both values (before and after) must not result in an "inverse" window, i.e., upper-interval bound cannot be smaller than lower-interval bound.
JoinWindows are sliding windows, thus, they are aligned to the actual record timestamps.
This implies, that each input record defines its own window with start and end time being relative to the record's
timestamp.
For time semantics, see TimestampExtractor.
TimeWindows,
UnlimitedWindows,
SessionWindows,
KStream.join(KStream, ValueJoiner, JoinWindows),
KStream.join(KStream, ValueJoiner, JoinWindows, StreamJoined),
KStream.leftJoin(KStream, ValueJoiner, JoinWindows),
KStream.leftJoin(KStream, ValueJoiner, JoinWindows, StreamJoined),
KStream.outerJoin(KStream, ValueJoiner, JoinWindows),
KStream.outerJoin(KStream, ValueJoiner, JoinWindows, StreamJoined),
TimestampExtractor| Modifier and Type | Field and Description |
|---|---|
long |
afterMs
Maximum time difference for tuples that are after the join tuple.
|
long |
beforeMs
Maximum time difference for tuples that are before the join tuple.
|
| Modifier and Type | Method and Description |
|---|---|
JoinWindows |
after(Duration timeDifference)
Changes the end window boundary to
timeDifference but keep the start window boundary as is. |
JoinWindows |
after(long timeDifferenceMs)
Deprecated.
Use
after(Duration) instead |
JoinWindows |
before(Duration timeDifference)
Changes the start window boundary to
timeDifference but keep the end window boundary as is. |
JoinWindows |
before(long timeDifferenceMs)
Deprecated.
Use
before(Duration) instead. |
boolean |
equals(Object o) |
JoinWindows |
grace(Duration afterWindowEnd)
Reject out-of-order events that are delayed more than
afterWindowEnd
after the end of its window. |
long |
gracePeriodMs()
Return the window grace period (the time to admit
out-of-order events after the end of the window.)
Delay is defined as (stream_time - record_timestamp).
|
int |
hashCode() |
long |
maintainMs()
Deprecated.
since 2.1. This function should not be used anymore, since
until(long)
is deprecated in favor of grace(Duration). |
static JoinWindows |
of(Duration timeDifference)
Specifies that records of the same key are joinable if their timestamps are within
timeDifference,
i.e., the timestamp of a record from the secondary stream is max timeDifference earlier or later than
the timestamp of the record from the primary stream. |
static JoinWindows |
of(long timeDifferenceMs)
Deprecated.
Use
of(Duration) instead. |
long |
size()
Return the size of the specified windows in milliseconds.
|
String |
toString() |
JoinWindows |
until(long durationMs)
Deprecated.
since 2.1. Use
grace(Duration) instead. |
Map<Long,Window> |
windowsFor(long timestamp)
Not supported by
JoinWindows. |
public final long beforeMs
public final long afterMs
@Deprecated public static JoinWindows of(long timeDifferenceMs) throws IllegalArgumentException
of(Duration) instead.timeDifferenceMs,
i.e., the timestamp of a record from the secondary stream is max timeDifferenceMs earlier or later than
the timestamp of the record from the primary stream.timeDifferenceMs - join window interval in millisecondsIllegalArgumentException - if timeDifferenceMs is negativepublic static JoinWindows of(Duration timeDifference) throws IllegalArgumentException
timeDifference,
i.e., the timestamp of a record from the secondary stream is max timeDifference earlier or later than
the timestamp of the record from the primary stream.timeDifference - join window intervalIllegalArgumentException - if timeDifference is negative or can't be represented as long milliseconds@Deprecated public JoinWindows before(long timeDifferenceMs) throws IllegalArgumentException
before(Duration) instead.timeDifferenceMs but keep the end window boundary as is.
Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
timeDifferenceMs earlier than the timestamp of the record from the primary stream.
timeDifferenceMs can be negative but its absolute value must not be larger than current window "after"
value (which would result in a negative window size).timeDifferenceMs - relative window start time in millisecondsIllegalArgumentException - if the resulting window size is negativepublic JoinWindows before(Duration timeDifference) throws IllegalArgumentException
timeDifference but keep the end window boundary as is.
Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
timeDifference earlier than the timestamp of the record from the primary stream.
timeDifference can be negative but its absolute value must not be larger than current window "after"
value (which would result in a negative window size).timeDifference - relative window start timeIllegalArgumentException - if the resulting window size is negative or timeDifference can't be represented as long milliseconds@Deprecated public JoinWindows after(long timeDifferenceMs) throws IllegalArgumentException
after(Duration) insteadtimeDifferenceMs but keep the start window boundary as is.
Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
timeDifferenceMs later than the timestamp of the record from the primary stream.
timeDifferenceMs can be negative but its absolute value must not be larger than current window "before"
value (which would result in a negative window size).timeDifferenceMs - relative window end time in millisecondsIllegalArgumentException - if the resulting window size is negativepublic JoinWindows after(Duration timeDifference) throws IllegalArgumentException
timeDifference but keep the start window boundary as is.
Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
timeDifference later than the timestamp of the record from the primary stream.
timeDifference can be negative but its absolute value must not be larger than current window "before"
value (which would result in a negative window size).timeDifference - relative window end timeIllegalArgumentException - if the resulting window size is negative or timeDifference can't be represented as long millisecondspublic Map<Long,Window> windowsFor(long timestamp)
JoinWindows.
Throws UnsupportedOperationException.windowsFor in class Windows<Window>timestamp - the timestamp window should get created forwindowStartTimestamp -> Window entriesUnsupportedOperationException - at every invocationpublic long size()
Windowspublic JoinWindows grace(Duration afterWindowEnd) throws IllegalArgumentException
afterWindowEnd
after the end of its window.
Delay is defined as (stream_time - record_timestamp).
afterWindowEnd - The grace period to admit out-of-order events to a window.IllegalArgumentException - if the afterWindowEnd is negative of can't be represented as long millisecondspublic long gracePeriodMs()
WindowsgracePeriodMs in class Windows<Window>@Deprecated public JoinWindows until(long durationMs) throws IllegalArgumentException
grace(Duration) instead.Windowsuntil in class Windows<Window>durationMs - the window retention time in millisecondsIllegalArgumentException - if durationMs is smaller than the window size@Deprecated public long maintainMs()
until(long)
is deprecated in favor of grace(Duration).
For TimeWindows the maintain duration is at least as small as the window size.
maintainMs in class Windows<Window>