Spark Struct Streaming - joins

2.3.0 부터 지원된다.,

(IPC - 2.2.0 , EPC 2.3.0)

Support matrix for joins in streaming queries

Left Input	Right Input	Join Type
Static	Static	All types	Supported, since its not on streaming data even though it can be present in a streaming query
Stream	Static	Inner	Supported, not stateful
		Left Outer	Supported, not stateful
		Right Outer	Not supported
		Full Outer	Not supported
Static	Stream	Inner	Supported, not stateful
		Left Outer	Not supported
		Right Outer	Supported, not stateful
		Full Outer	Not supported
Stream	Stream	Inner	Supported, optionally specify watermark on both sides + time constraints for state cleanup
		Left Outer	Conditionally supported, must specify watermark on right + time constraints for correct results, optionally specify watermark on left for all state cleanup
		Right Outer	Conditionally supported, must specify watermark on left + time constraints for correct results, optionally specify watermark on right for all state cleanup
		Full Outer	Not supported

import org.apache.spark.sql.functions.expr   val impressions = spark.readStream. ... val clicks = spark.readStream. ...   // Apply watermarks on event-time columns val impressionsWithWatermark = impressions.withWatermark("impressionTime", "2 hours") val clicksWithWatermark = clicks.withWatermark("clickTime", "3 hours")   // Join with event-time constraints impressionsWithWatermark.join(   clicksWithWatermark,   expr("""     clickAdId = impressionAdId AND     clickTime >= impressionTime AND     clickTime <= impressionTime + interval 1 hour     """) )     impressionsWithWatermark.join(   clicksWithWatermark,   expr("""     clickAdId = impressionAdId AND     clickTime >= impressionTime AND     clickTime <= impressionTime + interval 1 hour     """),   joinType = "leftOuter"      // can be "inner", "leftOuter", "rightOuter"  )

정리하자면, STREAM - STREAM JOIN 에서 INNER 조인은 watermark 가 옵셔날이지만 의미적으로도 써주는것이 좋고 성능도개선된다.

OUTER 조인은 Watermark 가 필수인데 ,그이유는 마지막까지 조인할 대상을 찾고 기다리기 때문에,  만약 null 일 경우 delay time 은 최대가된다.

• Joins can be cascaded, that is, you can do df1.join(df2, ...).join(df3, ...).join(df4, ....).

• As of Spark 2.3, you can use joins only when the query is in Append output mode. Other output modes are not yet supported.

• As of Spark 2.3, you cannot use other non-map-like operations before joins. Here are a few examples of what cannot be used.

  • Cannot use streaming aggregations before joins.

  • Cannot use mapGroupsWithState and flatMapGroupsWithState in Update mode before joins.