Stream Joins

Event yang mengalir di TokoKita jarang berdiri sendiri. Laporan penjualan butuh kategori dan harga yang tersimpan di tabel katalog, dan tim marketing butuh menghubungkan tayangan iklan dengan klik yang menyusul beberapa detik kemudian. Kedua kebutuhan tersebut adalah join — namun join batch yang kita praktikkan pada Chapter Spark Joins mengasumsikan kedua tabel sudah lengkap. Pada stream, satu sisi (atau keduanya) tidak pernah selesai tiba, sehingga pertanyaannya berubah: berapa lama engine harus menunggu pasangan, dan kapan penantuan itu boleh dihentikan? Chapter ini menjalankan dua pola join streaming secara nyata: stream-static untuk enrichment dan stream-stream berbasis watermark.

Stream-Static Join: Memperkaya Pesanan #

Sisi statis adalah tabel biasa yang isinya lengkap saat kueri berjalan — katalog produk, tabel dimensi pelanggan, daftar harga promo. Setiap micro-batch pesanan digabungkan dengan versi terbaru tabel tersebut, sehingga tidak ada state join yang perlu disimpan antar batch dan tidak ada watermark yang dibutuhkan.

python
"""Stream-static join: pesanan streaming diperkaya katalog produk statis."""
import time

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "false")
         .config("spark.sql.shuffle.partitions", "2")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

t0 = time.time()

# Sisi statis: katalog produk TokoKita (tabel dimensi kecil).
katalog = spark.createDataFrame(
    [
        ("Rok Tenun", "pakaian", 249000),
        ("Kain Songket", "pakaian", 375000),
        ("Tas Rotan", "aksesoris", 189000),
        ("Blouse Katun", "pakaian", 215000),
    ],
    "produk STRING, kategori STRING, harga INT",
)

# Sisi streaming: pesanan masuk dari aplikasi kasir (rate source).
daftar = F.array(F.lit("Rok Tenun"), F.lit("Kain Songket"),
                 F.lit("Tas Rotan"), F.lit("Blouse Katun"))
pesanan = (spark.readStream.format("rate")
           .option("rowsPerSecond", 2)
           .option("numPartitions", 1)
           .load()
           .withColumn("pesanan_id", (F.col("value") + 1000).cast("int"))
           .withColumn("produk", F.element_at(daftar, (F.col("value") % 4 + 1).cast("int")))
           .withColumn("qty", (F.col("value") % 3 + 1).cast("int")))

# Stream-static join: setiap micro-batch pesanan digabung katalog terbaru.
diperkaya = (pesanan.join(katalog, "produk", "inner")
             .select("pesanan_id", "produk", "qty", "kategori", "harga"))

q = (diperkaya.writeStream
     .format("console")
     .outputMode("append")
     .trigger(processingTime="5 seconds")
     .option("truncate", "false")
     .start())
q.awaitTermination(16)
q.stop()

print(f"durasi script: {time.time() - t0:.1f} detik")
spark.stop()

Output yang Dihasilkan (Batch 0 kosong; Batch 2 dan 3 berpola sama dengan Batch 1, hingga 10 baris per batch di kondisi stabil — dipangkas agar ringkas):

text
-------------------------------------------
Batch: 1
-------------------------------------------
+----------+------------+---+---------+------+
|pesanan_id|produk      |qty|kategori |harga |
+----------+------------+---+---------+------+
|1000      |Rok Tenun   |1  |pakaian  |249000|
|1001      |Kain Songket|2  |pakaian  |375000|
|1002      |Tas Rotan   |3  |aksesoris|189000|
|1003      |Blouse Katun|1  |pakaian  |215000|
|1004      |Rok Tenun   |2  |pakaian  |249000|
|1005      |Kain Songket|3  |pakaian  |375000|
+----------+------------+---+---------+------+
durasi script: 18.8 detik

Mari kita telaah baris-baris kode di atas:

Baris 17-25 (katalog = spark.createDataFrame(...)): sisi statis berupa DataFrame biasa — empat produk beserta kategori dan harga, meniru tabel dimensi di warehouse.
Baris 30-36 (pesanan = ...): sisi streaming dari rate source; setiap baris diberi pesanan_id, nama produk bergilir, dan qty.
Baris 39-40 (pesanan.join(katalog, "produk", "inner")): inti pola ini — join memakai sintaks yang persis sama dengan join batch, tanpa watermark, karena katalog selalu tersedia lengkap.
Baris 42-48 (writeStream): hasil join ditulis dengan output mode append karena setiap baris hasil bersifat final begitu pasangannya ditemukan.

Setiap pesanan keluar sudah membawa kategori dan harga yang tidak pernah dikirim oleh kasir — enrichment selesai di dalam pipeline, bukan di dashboard. Jumlah baris batch pertama (6) lebih kecil daripada kondisi stabil (10 baris = 2 baris/detik × 5 detik trigger) karena startup JVM dan sumber rate belum sepenuhnya jalan; batch berikutnya menyusul hingga skrip berhenti setelah 16 detik.

Stream-Stream Join: Interval Join Tayangan dan Klik #

Dua stream yang keduanya belum selesai menimbulkan masalah yang tidak ada di batch: untuk setiap tayangan iklan, engine harus menyimpan baris itu di state sambil menunggu klik yang mungkin datang — dan tanpa batas waktu, penantuan itu tidak pernah berakhir. Solusinya adalah kombinasi watermark di kedua sisi dan syarat batas waktu pada event time.

python
"""Stream-stream join: tayangan iklan digabung klik dalam rentang event time."""
import time

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "false")
         .config("spark.sql.shuffle.partitions", "2")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

t0 = time.time()

# Stream A: tayangan iklan (event_time = waktu kedatangan), 1 baris/detik.
tayangan = (spark.readStream.format("rate")
            .option("rowsPerSecond", 1)
            .option("numPartitions", 1)
            .load()
            .withColumn("iklan_id", (F.col("value") + 1).cast("int"))
            .withColumn("event_time", F.col("timestamp"))
            .withWatermark("event_time", "5 seconds"))

# Stream B: klik iklan (event_time 4 detik setelah tayangan bernilai sama).
klik = (spark.readStream.format("rate")
        .option("rowsPerSecond", 1)
        .option("numPartitions", 1)
        .load()
        .withColumn("iklan_id", (F.col("value") + 1).cast("int"))
        .withColumn("event_time", F.col("timestamp") + F.expr("INTERVAL 4 seconds"))
        .withWatermark("event_time", "5 seconds"))

# Interval join: klik harus jatuh antara tayangan dan 10 detik setelahnya.
hasil = (tayangan.alias("t")
         .join(klik.alias("k"),
               F.expr("k.iklan_id = t.iklan_id AND "
                      "k.event_time BETWEEN t.event_time "
                      "AND t.event_time + INTERVAL 10 seconds"))
         .select(F.col("t.iklan_id"),
                 F.col("t.event_time").alias("t_tayang"),
                 F.col("k.event_time").alias("t_klik"),
                 (F.col("k.event_time") - F.col("t.event_time")).alias("selisih")))

q = (hasil.writeStream
     .format("console")
     .outputMode("append")
     .trigger(processingTime="5 seconds")
     .option("truncate", "false")
     .start())
q.awaitTermination(30)
q.stop()

print(f"durasi script: {time.time() - t0:.1f} detik")
spark.stop()

Output yang Dihasilkan (Batch 0 kosong; Batch 2 sampai 6 berpola sama dengan pasangan iklan_id berikutnya — dipangkas agar ringkas):

text
-------------------------------------------
Batch: 1
-------------------------------------------
+--------+-----------------------+-----------------------+---------------------------------------+
|iklan_id|t_tayang               |t_klik                 |selisih                                |
+--------+-----------------------+-----------------------+---------------------------------------+
|2       |2026-10-16 14:07:17.719|2026-10-16 14:07:21.753|INTERVAL '0 00:00:04.034' DAY TO SECOND|
|1       |2026-10-16 14:07:16.719|2026-10-16 14:07:20.753|INTERVAL '0 00:00:04.034' DAY TO SECOND|
|3       |2026-10-16 14:07:18.719|2026-10-16 14:07:22.753|INTERVAL '0 00:00:04.034' DAY TO SECOND|
+--------+-----------------------+-----------------------+---------------------------------------+
durasi script: 32.9 detik

Baris 17-23 (tayangan = ...): stream tayangan diberi iklan_id dan watermark 5 detik — engine tahu tayangan yang lebih tua dari watermark tidak akan datang lagi.
Baris 26-32 (klik = ...): stream klik memodelkan pengguna yang menunggu sekitar 4 detik sebelum mengeklik; sisi ini juga wajib diberi watermark.
Baris 35-39 (F.expr("k.iklan_id = t.iklan_id AND k.event_time BETWEEN ...")): syarat join memuat batas event time — klik sah hanya jika jatuh di antara tayangan dan 10 detik setelahnya. Batas inilah yang memberi tahu Spark: begitu watermark melewati t.event_time + 10 detik, pasangan tayangan itu tidak akan pernah datang dan state-nya boleh dihapus.
Baris 40-43 (select ... alias("selisih")): kolom selisih memperlihatkan jarak tayang-ke-klik, tercatat 4,034 detik — 4 detik jeda yang dimodelkan plus 34 milidetik offset antar dua sumber rate.

Pasangan tayangan-klik keluar begitu kedua sisinya bertemu, dan watermark membersihkan baris tayangan yang tidak pernah diklik setelah batas 10 detik terlampaui. Timestamp dan selisih bervariasi antar run karena bergantung jam dinding dan jeda milidetik kedua sumber; pola join-nya stabil. Perhatikan juga urutan iklan_id yang acak di output — Spark menghasilkan baris per partition tanpa janji urutan, sama seperti hasil join batch.

Aturan Main Join Streaming #

Bentuk join	Watermark	Syarat tambahan
Stream-static, semua tipe	Tidak wajib	Sisi statis dibaca ulang setiap micro-batch
Stream-stream, inner	Dianjurkan kedua sisi	Batas waktu antar event membatasi state
Stream-stream, left/right/full outer	Wajib kedua sisi	Batas waktu wajib; hasil outer tertunda sampai watermark melewati batas

Tiga baris tabel tersebut merangkum satu prinsip: makin tidak pasti kedua sisi, makin banyak syarat waktu yang harus dideklarasikan. Rincian lengkap batasan tiap kombinasi terdokumentasi di Structured Streaming Programming Guide .

Hasil join yang mengalir ke hilir tetap bergantung pada satu asumsi diam-diam: kedua sisi sepakat tentang bentuk data — nama field dan tipenya. Saat tim kasir mengubah field harga menjadi price, join bisa rusak tanpa pesan galat. Menjaga kontrak bentuk data antar tim adalah pokok bahasan Chapter Schema Registry.