Streaming joins in Apache Spark

Introduction

Streaming joins let you combine data from two live streams as new data arrives. This helps you analyze related events happening in real time.

Matching user clicks with ad impressions as they happen on a website.

Joining sensor data streams from two machines to detect correlated faults.

Combining live stock trades with real-time news feeds to spot market trends.

Merging live GPS data from vehicles with traffic updates to optimize routes.

Syntax

Apache Spark

stream1.join(stream2, join_expression, join_type)

stream1 and stream2 are streaming DataFrames.

join_expression defines how to match rows (like keys).

Examples

Inner join streams on matching keys, keeping only matching rows.

Apache Spark

stream1.join(stream2, stream1.key == stream2.key, 'inner')

Left outer join keeps all rows from stream1 and matches from stream2 if available.

Apache Spark

stream1.join(stream2, stream1.id == stream2.id, 'left_outer')

Right outer join keeps all rows from stream2 and matches from stream1 if available.

Apache Spark

stream1.join(stream2, expr("stream1.key == stream2.key"), 'right_outer')

Sample Program

This code creates two streaming sources that generate numbers and timestamps. It renames columns to create keys, then joins the streams where keys match. The result prints to the console for 5 seconds.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr

spark = SparkSession.builder.appName('StreamingJoinsExample').getOrCreate()

# Simulate streaming data with rate source
stream1 = spark.readStream.format('rate').option('rowsPerSecond', 5).load()
stream2 = spark.readStream.format('rate').option('rowsPerSecond', 5).load()

# Rename columns to simulate keys
stream1 = stream1.selectExpr('value as key1', 'timestamp as ts1')
stream2 = stream2.selectExpr('value as key2', 'timestamp as ts2')

# Join streams on keys where key1 == key2
joined_stream = stream1.join(stream2, expr('key1 == key2'), 'inner')

query = joined_stream.writeStream.format('console').outputMode('append').start()

query.awaitTermination(5000)  # Run for 5 seconds then stop
query.stop()

Important Notes

Streaming joins require careful handling of late or missing data to avoid memory issues.

Use watermarking to limit how long Spark waits for matching data in streams.

Not all join types are supported in streaming; inner and some outer joins are common.

Summary

Streaming joins combine two live data streams based on matching keys.

They help analyze related events happening at the same time.

Use Spark's join syntax with streaming DataFrames and manage late data with watermarks.