Structured Streaming 심화

from pyspark.sql.functions import window

# 5분 단위 고정 윈도우
df.withWatermark("event_time", "10 minutes") \
  .groupBy(window("event_time", "5 minutes"), "device_id") \
  .agg(avg("temperature").alias("avg_temp"))

# 결과 윈도우: [10:00-10:05), [10:05-10:10), [10:10-10:15), ...

# 10분 윈도우, 5분마다 슬라이딩
df.withWatermark("event_time", "10 minutes") \
  .groupBy(window("event_time", "10 minutes", "5 minutes"), "device_id") \
  .agg(avg("temperature").alias("avg_temp"))

# 결과 윈도우: [10:00-10:10), [10:05-10:15), [10:10-10:20), ...
# → 각 이벤트가 2개 윈도우에 포함됨

from pyspark.sql.functions import session_window

# 30분 이상 활동이 없으면 세션 종료
df.withWatermark("event_time", "1 hour") \
  .groupBy(session_window("event_time", "30 minutes"), "user_id") \
  .agg(
      count("*").alias("page_views"),
      min("event_time").alias("session_start"),
      max("event_time").alias("session_end")
  )

# Append Mode: 새 행만 쓰기 (가장 일반적)
df.writeStream.outputMode("append").toTable("output")

# Update Mode: 변경된 행만 쓰기
df.groupBy("device_id").agg(avg("temp")) \
  .writeStream.outputMode("update").toTable("device_avg")

# Complete Mode: 전체 결과 덮어쓰기 (소규모만)
df.groupBy("status").count() \
  .writeStream.outputMode("complete") \
  .format("memory").queryName("status_count").start()

-- 테이블에 CDF 활성화
ALTER TABLE catalog.schema.customers
SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

-- 새 테이블 생성 시 활성화
CREATE TABLE catalog.schema.orders (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    status STRING
)
TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

# CDF가 활성화된 Delta 테이블에서 변경사항 읽기
changes_df = (
    spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")      # CDF 활성화
    .option("startingVersion", 0)           # 또는 "startingTimestamp"
    .table("catalog.schema.customers")
)

# CDF 메타데이터 컬럼
# _change_type: "insert", "update_preimage", "update_postimage", "delete"
# _commit_version: 변경이 발생한 Delta 버전
# _commit_timestamp: 변경 시각

changes_df.select(
    "_change_type",
    "_commit_version",
    "customer_id",
    "name",
    "email"
).display()

# CDF로 SCD Type 2 구현 (foreachBatch 패턴)
def upsert_to_scd2(batch_df, batch_id):
    from delta.tables import DeltaTable

    # UPDATE: 기존 행의 end_date를 설정 (이력 닫기)
    updates = batch_df.filter("_change_type = 'update_postimage'")

    target = DeltaTable.forName(spark, "catalog.schema.customers_scd2")

    target.alias("t").merge(
        updates.alias("s"),
        "t.customer_id = s.customer_id AND t.is_current = true"
    ).whenMatchedUpdate(set={
        "is_current": "false",
        "end_date": "s._commit_timestamp"
    }).whenNotMatchedInsert(values={
        "customer_id": "s.customer_id",
        "name": "s.name",
        "email": "s.email",
        "start_date": "s._commit_timestamp",
        "end_date": "null",
        "is_current": "true"
    }).execute()

# 스트림 실행
(changes_df
    .writeStream
    .foreachBatch(upsert_to_scd2)
    .option("checkpointLocation", "/checkpoints/scd2")
    .trigger(availableNow=True)
    .start()
)

# StreamingQueryListener로 상태 크기 모니터링
query = df.writeStream...start()

# 진행 상태에서 State Store 메트릭 확인
progress = query.lastProgress
if progress and progress.get("stateOperators"):
    for op in progress["stateOperators"]:
        print(f"State rows: {op['numRowsTotal']:,}")
        print(f"State memory: {op['memoryUsedBytes'] / 1024 / 1024:.1f} MB")
        print(f"Rows dropped (late): {op.get('numRowsDroppedByWatermark', 0):,}")