Spark Structured Streaming

# Kafka에서 스트림 읽기
df_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
    .option("subscribe", "my-topic")
    .option("startingOffsets", "latest")
    .load()
)

# Delta 테이블에서 스트림 읽기 (CDC 등)
df_delta_stream = (
    spark.readStream
    .format("delta")
    .table("catalog.schema.source_table")
)

# Auto Loader로 클라우드 파일 스트림 읽기
df_autoloader = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/schema")
    .load("s3://my-bucket/raw-data/")
)

# Delta 테이블로 스트림 쓰기
query = (
    df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.target_table")
)

# 30초마다 마이크로 배치 실행
query = (
    df_stream.writeStream
    .format("delta")
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", "/checkpoints/stream1")
    .toTable("catalog.schema.output")
)

# 가용한 데이터 모두 처리 후 종료 (스케줄링된 잡에 적합)
query = (
    df_stream.writeStream
    .format("delta")
    .trigger(availableNow=True)
    .option("checkpointLocation", "/checkpoints/stream2")
    .toTable("catalog.schema.output")
)

from pyspark.sql.functions import window, col

# 이벤트 타임 기준, 최대 10분 늦은 데이터까지 허용
df_windowed = (
    df_stream
    .withWatermark("event_time", "10 minutes")
    .groupBy(
        window("event_time", "5 minutes"),  # 5분 윈도우
        "device_id"
    )
    .agg({"temperature": "avg", "humidity": "avg"})
)

현재 최대 이벤트 시간: 10:30
워터마크 임계값:      10분
워터마크:            10:20

→ 10:20 이전 이벤트 타임을 가진 데이터는 '너무 늦음'으로 판단하여 무시
→ 10:20 이후 데이터는 정상 처리

# 스트림: 주문 이벤트
orders_stream = spark.readStream.table("catalog.schema.orders_raw")

# 스태틱: 고객 정보 (차원 테이블)
customers = spark.read.table("catalog.schema.customers")

# 조인: 주문에 고객 정보 추가
enriched_orders = orders_stream.join(
    customers,
    orders_stream.customer_id == customers.customer_id,
    "left"
)

# 스트림 1: 광고 노출
impressions = (
    spark.readStream
    .table("catalog.schema.impressions")
    .withWatermark("impression_time", "2 hours")
)

# 스트림 2: 광고 클릭
clicks = (
    spark.readStream
    .table("catalog.schema.clicks")
    .withWatermark("click_time", "3 hours")
)

# 조인: 노출 후 1시간 이내에 발생한 클릭 매칭
matched = impressions.join(
    clicks,
    (impressions.ad_id == clicks.ad_id) &
    (clicks.click_time >= impressions.impression_time) &
    (clicks.click_time <= impressions.impression_time + expr("INTERVAL 1 HOUR")),
    "leftOuter"
)

query = (
    df_stream.writeStream
    .format("delta")
    .option("checkpointLocation", "/mnt/checkpoints/my-pipeline/stream1")
    .toTable("catalog.schema.output")
)

# Kafka 연동 예시
df_kafka = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "earliest")
    .load()
)

# Kafka value를 JSON으로 파싱
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("order_id", StringType()) \
    .add("amount", DoubleType()) \
    .add("timestamp", StringType())

df_parsed = df_kafka.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

def upsert_to_delta(batch_df, batch_id):
    from delta.tables import DeltaTable
    target = DeltaTable.forName(spark, "catalog.schema.customers")
    (target.alias("t")
        .merge(batch_df.alias("s"), "t.customer_id = s.customer_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())

query = (
    df_stream.writeStream
    .foreachBatch(upsert_to_delta)
    .option("checkpointLocation", "/checkpoints/upsert-stream")
    .trigger(processingTime="1 minute")
    .start()
)

# 실행 중인 스트리밍 쿼리 상태 확인
for q in spark.streams.active:
    print(f"쿼리: {q.name}, 상태: {q.status}")

# 마지막 마이크로 배치 진행 정보
query.lastProgress