🧱 Structured Streaming

Потоковые запросы, режимы триггеров, Kafka, watermarking, checkpointing

Databricks Data Engineer · DE-Associate · ⏱️ 22 min · ❓ 4 вопр.

🌊 Что такое Structured Streaming?

Spark Structured Streaming — непрерывная обработка DataFrame. Модель unbounded table: новые данные = новые строки. Похоже на Batch API: `spark.readStream`, `writeStream`. Fault tolerant: через checkpoint. Micro-batch и continuous processing.

⏱️ Режимы триггеров (Trigger Modes)

Trigger.ProcessingTime("1 minute") — micro-batch каждую минуту. Trigger.Once() — однократный запуск (deprecated). Trigger.AvailableNow() — полностью считывает доступные данные и завершается (вместо Once). Trigger.Continuous("1 second") — настоящий стриминг, задержка ~мс, пока экспериментальный. По умолчанию: запускается сразу, как только возможно.

🚰 Интеграция с Kafka

`spark.readStream.format("kafka")` — чтение потока из topic. Key, value, topic, partition, offset, timestamp — стандартные поля Kafka. Value в формате `binary` — декодируется через `.cast("string")`. `startingOffsets` — "earliest", "latest" или JSON offset. `subscribe` (несколько topics) и `subscribePattern` (regex).

💧 Watermarking и stateful-операции

Watermark — допуск для запоздавших данных. `.withWatermark("event_time", "10 minutes")` — принимаются данные с опозданием до 10 минут. Stateful operations: windowed aggregation, stream-stream join. Режимы вывода (Output modes): Append (новые строки), Complete (все строки), Update (изменённые строки).

💡 Ключевые моменты

readStream/writeStream — API потокового DataFrame
Checkpoint: необходим для exactly-once и fault tolerance
Триггер AvailableNow: работает как batch, вместо Once (deprecated)
Watermark: допуск для запоздавших событий
Формат Kafka: key, value, topic, partition, offset
Output modes: Append, Complete, Update

📋 Пример кода

# 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")
)

🎯 Советы для экзамена

Trigger.Once() устарел (deprecated) — следует использовать `AvailableNow()`
checkpointLocation — обязателен в writeStream (fault tolerance)
Значение Kafka в binary — декодируется через `F.col("value").cast("string")`
Watermark + windowed aggregation: группировка по окну event_time
Режим Append — не совместим с агрегацией (нужен Update или Complete)

⚠️ Частые ошибки

Считать, что exactly-once гарантируется без checkpoint — checkpoint обязателен
Использовать Trigger.Once() — устарел, нужно использовать AvailableNow()
Считать значение Kafka напрямую строкой — оно binary, требуется cast

🧠 Для запоминания: "Stream = Река": Source (Kafka/S3) → Transform (filter/agg) → Sink (Delta/Kafka). "CAUW" = Checkpoint, AvailableNow, Update-mode, Watermark — 4 ключевых понятия стриминга

Проверьте себя по теме «Structured Streaming»

Бесплатные интерактивные тесты, пробный экзамен и полные уроки — на платформе CertMaster.

Начать бесплатно →