🧱 Structured Streaming

Streaming queries, trigger modes, Kafka, watermarking, checkpointing

Databricks Data Engineer · DE-Associate · ⏱️ 22 min · ❓ 4 savol

🌊 Structured Streaming nima?

Spark Structured Streaming — continuous DataFrame processing. Unbounded table model: yangi data = yangi qatorlar. Batch API ga o'xshash: `spark.readStream`, `writeStream`. Fault tolerant: checkpoint orqali. Micro-batch va continuous processing.

⏱️ Trigger Modes

Trigger.ProcessingTime("1 minute") — har daqiqada micro-batch. Trigger.Once() — bir marta run (deprecated). Trigger.AvailableNow() — mavjud data ni to'liq o'qib tugaydi (Once o'rniga). Trigger.Continuous("1 second") — true streaming, ~ms latency, hali experimental. Default: mümkin bo'lganda darhol.

🚰 Kafka Integration

`spark.readStream.format("kafka")` — topic dan stream o'qish. Key, value, topic, partition, offset, timestamp — standart Kafka fields. Value `binary` — `.cast("string")` bilan decode. `startingOffsets` — "earliest", "latest", yoki JSON offset. `subscribe` (ko'p topic) va `subscribePattern` (regex).

💧 Watermarking & Stateful Operations

Watermark — kechikkan data uchun toleransiya. `.withWatermark("event_time", "10 minutes")` — 10 minutgacha kechikkan data qabul qilinadi. Stateful operations: windowed aggregation, stream-stream join. Output modes: Append (yangi qatorlar), Complete (barcha), Update (o'zgargan).

💡 Asosiy nuqtalar

readStream/writeStream — streaming DataFrame API
Checkpoint: exactly-once, fault tolerance uchun zarur
AvailableNow trigger: batch-like, Once o'rniga (deprecated)
Watermark: kechikkan event uchun toleransiya
Kafka format: key, value, topic, partition, offset
Output modes: Append, Complete, Update

📋 Kod misoli

# Kafka dan stream o'qish
df_stream = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "orders")
  .option("startingOffsets", "latest")
  .load()
)

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StringType

# JSON parse
schema = StructType().add("id", StringType()).add("amount", "double")

orders = (df_stream
  .select(F.from_json(F.col("value").cast("string"), schema).alias("data"))
  .select("data.*")
  .withWatermark("event_time", "5 minutes")
)

# Delta Lake ga yozish
(orders.writeStream
  .format("delta")
  .outputMode("append")
  .option("checkpointLocation", "/checkpoints/orders")
  .trigger(availableNow=True)  # AvailableNow
  .table("silver.orders")
)

🎯 Imtihon maslahatlari

Trigger.Once() deprecated — `AvailableNow()` ishlatish kerak
checkpointLocation — writeStream da majburiy (fault tolerance)
Kafka value binary — `F.col("value").cast("string")` bilan decode
Watermark + windowed aggregation: event_time oynasida group qilish
Append mode — aggregation bilan ishlatib bo'lmaydi (Update yoki Complete kerak)

⚠️ Ko'p adashadigan

Checkpoint yo'qligida exactly-once kafolatlanadi deb o'ylash — checkpoint majburiy
Trigger.Once() ishlatish — deprecated, AvailableNow() ishlatish kerak
Kafka value directly string deb hisoblash — binary, cast kerak

🧠 Eslab qolish: "Stream = Daryo": Source (Kafka/S3) → Transform (filter/agg) → Sink (Delta/Kafka). "CAUW" = Checkpoint, AvailableNow, Update-mode, Watermark — 4 muhim streaming tushuncha