7장: Spark Structured Streaming

입력 테이블 (끊임없이 행 추가)
┌─────────┬────────┬────────┐
│ time    │ user   │ amount │
├─────────┼────────┼────────┤
│ 09:00   │ Kim    │ 1000   │  ← 기존 데이터
│ 09:01   │ Lee    │ 2000   │
│ 09:02   │ Park   │ 1500   │  ← 새 데이터 (이번 트리거)
│ 09:03   │ Kim    │ 3000   │
└─────────┴────────┴────────┘
            ↓ 증분 쿼리 실행
결과 테이블
┌────────┬──────────────┐
│ user   │ total_amount │
├────────┼──────────────┤
│ Kim    │ 4000         │  ← 갱신됨
│ Lee    │ 2000         │
│ Park   │ 1500         │  ← 새로 추가
└────────┴──────────────┘

시간: ──|──100ms──|──100ms──|──100ms──|──
배치:    [B1      ] [B2      ] [B3      ]
이벤트:  ** *  *    *  **      * * *

query = (df.writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "output")
    .trigger(continuous="1 second")  # 연속 처리, 1초 체크포인트 간격
    .start())

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
 
spark = SparkSession.builder \
    .appName("OrderAnalytics") \
    .getOrCreate()
 
# 스키마 정의
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("category", StringType()),
    StructField("order_time", TimestampType())
])
 
# Kafka 소스에서 읽기
raw_df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "latest")
    .load())
 
# JSON 파싱
orders_df = (raw_df
    .select(from_json(col("value").cast("string"), order_schema).alias("data"))
    .select("data.*"))
 
# 처리: 카테고리별 5분 매출 집계
revenue_df = (orders_df
    .withWatermark("order_time", "5 minutes")
    .groupBy(
        window(col("order_time"), "5 minutes"),
        col("category")
    )
    .agg(spark_sum("amount").alias("total_revenue")))
 
# 결과 출력
query = (revenue_df.writeStream
    .outputMode("update")
    .format("console")
    .trigger(processingTime="10 seconds")
    .start())
 
query.awaitTermination()

from pyspark.sql.functions import window, count, avg
 
# 텀블링 윈도우: 10분 단위 집계
tumbling = (orders_df
    .withWatermark("order_time", "10 minutes")
    .groupBy(
        window(col("order_time"), "10 minutes"),
        col("category")
    )
    .agg(
        count("*").alias("order_count"),
        avg("amount").alias("avg_amount")
    ))
 
# 슬라이딩 윈도우: 10분 윈도우, 5분 슬라이드
sliding = (orders_df
    .withWatermark("order_time", "10 minutes")
    .groupBy(
        window(col("order_time"), "10 minutes", "5 minutes"),
        col("category")
    )
    .agg(count("*").alias("order_count")))

# 최대 10분까지 늦은 이벤트 허용
df_with_watermark = orders_df.withWatermark("order_time", "10 minutes")

def process_batch(batch_df, batch_id):
    # 배치 DataFrame에 대해 임의의 처리 가능
 
    # 1. Elasticsearch에 쓰기
    (batch_df.write
        .format("org.elasticsearch.spark.sql")
        .option("es.resource", "orders/_doc")
        .mode("append")
        .save())
 
    # 2. PostgreSQL에 upsert
    (batch_df.write
        .format("jdbc")
        .option("url", "jdbc:postgresql://postgres:5432/analytics")
        .option("dbtable", "order_summary")
        .option("user", "writer")
        .mode("append")
        .save())
 
    # 3. 캐시 갱신
    update_redis_cache(batch_df)
 
query = (revenue_df.writeStream
    .foreachBatch(process_batch)
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", "/checkpoints/revenue")
    .start())

def idempotent_process(batch_df, batch_id):
    # batch_id를 활용하여 중복 처리 방지
    if is_already_processed(batch_id):
        return
 
    # 처리 수행
    write_to_sink(batch_df)
 
    # 처리 완료 기록
    mark_as_processed(batch_id)

kafka_df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders,payments")       # 여러 토픽 구독
    .option("startingOffsets", "earliest")         # 처음부터 읽기
    .option("maxOffsetsPerTrigger", "10000")       # 트리거당 최대 오프셋
    .option("kafka.security.protocol", "SASL_SSL") # 보안 설정
    .load())
 
# Kafka 레코드 구조
# key: binary, value: binary, topic: string,
# partition: int, offset: long, timestamp: timestamp

# 처리 결과를 다른 Kafka 토픽에 쓰기
query = (processed_df
    .selectExpr("CAST(order_id AS STRING) AS key",
                "to_json(struct(*)) AS value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "orders-processed")
    .option("checkpointLocation", "/checkpoints/orders-processed")
    .trigger(processingTime="10 seconds")
    .start())