Spark Structured Streaming Scaling Strategy วิธี

Spark Structured Streaming คืออะไรและทำงานอย่างไร

Spark Structured Streaming เป็น stream processing engine ที่สร้างบน Spark SQL engine ใช้แนวคิด micro-batch processing ที่แบ่ง streaming data เป็น batch เล็กๆแล้วประมวลผลด้วย Spark SQL ทำให้สามารถใช้ DataFrame/Dataset API เดียวกันสำหรับทั้ง batch และ streaming workloads

ตั้งแต่ Spark 2.3 เพิ่ม Continuous Processing mode ที่ประมวลผลแบบ record-by-record ได้ latency ต่ำถึง 1 millisecond แต่ยังเป็น experimental อยู่ ส่วน micro-batch mode ที่เป็นค่า default มี latency ประมาณ 100ms-หลายวินาทีแต่มี throughput สูงกว่าและ fault tolerance ดีกว่า

Structured Streaming รองรับ source หลายประเภทเช่น Kafka สำหรับ message streaming, File source สำหรับอ่านไฟล์ใหม่ที่ถูกเพิ่มเข้ามา, Socket source สำหรับทดสอบ และ Rate source สำหรับ generate ข้อมูลทดสอบ ส่วน sink รองรับ Kafka, File, Console, Memory และ ForeachBatch สำหรับ custom output

การ scale Structured Streaming ต้องพิจารณาหลายปัจจัยเช่น จำนวน partitions ของ Kafka topic, จำนวน executors และ cores, memory configuration, watermark สำหรับ late data handling และ checkpoint strategy สำหรับ fault tolerance

เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ Elixir Phoenix LiveView SaaS Architecture

ตั้งค่า Spark Structured Streaming เบื้องต้น

สร้าง Streaming Application ที่อ่านจาก Kafka และเขียนไปยัง Delta Lake

# submit_streaming_job.sh — Submit Spark Streaming Job
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name "kafka-to-delta-streaming" \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 8g \
  --driver-memory 4g \
  --conf spark.sql.shuffle.partitions=200 \
  --conf spark.streaming.kafka.maxRatePerPartition=10000 \
  --conf spark.sql.streaming.checkpointLocation=s3a://bucket/checkpoints/ \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=5 \
  --conf spark.dynamicAllocation.maxExecutors=50 \
  --conf spark.sql.adaptive.enabled=true \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0, io.delta:delta-spark_2.12:3.1.0 \
  streaming_app.py

# streaming_app.py — Spark Structured Streaming Application
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("KafkaToDeltaStreaming") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Schema สำหรับ Kafka messages
event_schema = StructType([
    StructField("event_id", StringType()),
    StructField("user_id", StringType()),
    StructField("event_type", StringType()),
    StructField("timestamp", TimestampType()),
    StructField("properties", MapType(StringType(), StringType())),
])

# อ่านจาก Kafka
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka1:9092, kafka2:9092, kafka3:9092") \
    .option("subscribe", "user-events") \
    .option("startingOffsets", "latest") \
    .option("maxOffsetsPerTrigger", 1000000) \
    .option("kafka.consumer.group.id", "spark-streaming-group") \
    .option("failOnDataLoss", "false") \
    .load()

# Parse JSON messages
events_df = kafka_df \
    .select(from_json(col("value").cast("string"), event_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("timestamp", "10 minutes")

# Aggregate: นับ events ต่อ user ทุก 5 นาที
user_counts = events_df \
    .groupBy(
        window("timestamp", "5 minutes"),
        "user_id",
        "event_type"
    ) \
    .count()

# เขียนไป Delta Lake
query = user_counts.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3a://bucket/checkpoints/user-counts") \
    .option("mergeSchema", "true") \
    .trigger(processingTime="30 seconds") \
    .start("s3a://bucket/delta/user-event-counts")

query.awaitTermination()

กลยุทธ์ Scaling สำหรับ Streaming Workloads

วิธี scale Spark Structured Streaming ให้รองรับ throughput สูง

แนะนำเพิ่มเติม — บทวิเคราะห์จาก XM Signal

# กลยุทธ์ Scaling Spark Structured Streaming
#
# === 1. Horizontal Scaling (เพิ่ม Executors) ===
# กฎ: 1 Kafka partition = 1 Spark task
# ถ้า Kafka topic มี 100 partitions → ต้องมีอย่างน้อย 100 cores
#
# spark.executor.instances = ceil(kafka_partitions / executor_cores)
# ตัวอย่าง: 100 partitions, 4 cores/executor → 25 executors
#
# === 2. Dynamic Allocation ===
--conf spark.dynamicAllocation.enabled=true
--conf spark.dynamicAllocation.minExecutors=5
--conf spark.dynamicAllocation.maxExecutors=100
--conf spark.dynamicAllocation.executorIdleTimeout=60s
--conf spark.dynamicAllocation.schedulerBacklogTimeout=1s

# === 3. Kafka Partition Scaling ===
# เพิ่ม partitions ของ Kafka topic
kafka-topics.sh --alter --topic user-events \
  --partitions 200 \
  --bootstrap-server kafka1:9092

# === 4. Trigger Interval Tuning ===
# ลด trigger interval สำหรับ low latency
.trigger(processingTime="10 seconds")  # micro-batch ทุก 10 วินาที
# หรือใช้ available-now สำหรับ backfill
.trigger(availableNow=True)

# === 5. Shuffle Partitions ===
# กฎ: shuffle partitions ≈ 2-3x จำนวน cores
spark.conf.set("spark.sql.shuffle.partitions", 400)
# หรือใช้ AQE (Adaptive Query Execution)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# === 6. Memory Tuning ===
--executor-memory 8g
--conf spark.memory.fraction=0.6
--conf spark.memory.storageFraction=0.5
--conf spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
# RocksDB สำหรับ stateful streaming ที่ state ใหญ่มาก

# === 7. maxOffsetsPerTrigger ===
# จำกัด records ต่อ trigger เพื่อป้องกัน OOM
.option("maxOffsetsPerTrigger", 500000)

# === 8. Backpressure ===
spark.conf.set("spark.streaming.backpressure.enabled", "true")
spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "10000")

Tuning Performance และ Resource Optimization

ปรับแต่ง performance สำหรับ production streaming

Monitoring และ Alerting สำหรับ Streaming Jobs

สร้างระบบ monitoring สำหรับ Spark Streaming

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง GIT Remote คืออะไร — คู่มือโปรแกรมมิ่ง 2026

Fault Tolerance และ Exactly-Once Processing

ตั้งค่า checkpoint และ exactly-once semantics

FAQ คำถามที่พบบ่อย

Q: Spark Structured Streaming กับ Apache Flink ต่างกันอย่างไร?

A: Spark ใช้ micro-batch เป็นหลัก latency ประมาณ 100ms-วินาที เหมาะสำหรับ throughput สูง ใช้ DataFrame API เดียวกับ batch ส่วน Flink เป็น true streaming ที่ประมวลผลทีละ record latency ต่ำกว่า 10ms เหมาะสำหรับ low-latency use cases ถ้าต้องการ unified batch+streaming platform เลือก Spark ถ้าต้องการ low-latency เลือก Flink

แนะนำเพิ่มเติม — ระบบเทรดของ iCafeForex

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง SonarQube Analysis Production Setup Guide

Q: State store ใหญ่เกินไปทำอย่างไร?

A: ใช้ RocksDB state store provider แทน default in-memory store เพราะเก็บ state บน disk ได้ ตั้ง watermark ให้เหมาะสมเพื่อลบ state ที่เก่าเกินไป ใช้ mapGroupsWithState แทน groupBy เพื่อควบคุม state lifecycle เอง และพิจารณา state store cleanup ด้วย spark.sql.streaming.stateStore.maintenanceInterval

Q: maxOffsetsPerTrigger ควรตั้งเท่าไหร่?

A: ขึ้นอยู่กับ processing capacity ของ cluster เริ่มจากตั้งให้ batch ใช้เวลาประมาณ 50-70% ของ trigger interval เช่น trigger ทุก 30 วินาที batch ควรเสร็จใน 15-20 วินาที ถ้า batch ใช้เวลาเกิน trigger interval แสดงว่า maxOffsetsPerTrigger สูงเกินไป หรือต้องเพิ่ม resources

เนื้อหาเกี่ยวข้อง — อ่านต่อ: OAuth 2.1 Kubernetes Deployment

Q: จะ scale Kafka partitions โดยไม่หยุด streaming job ได้ไหม?

A: ได้ Spark Structured Streaming ตรวจจับ partitions ใหม่อัตโนมัติ แต่ต้อง restart streaming query เพื่อให้ consumer rebalance ใน production ใช้ rolling restart โดย stop query ด้วย query.stop() แล้ว start ใหม่ checkpoint จะ resume จาก offset ล่าสุดโดยอัตโนมัติ