Structured Streaming

1เตรียม Socket Server จำลอง

# เปิด terminal ใหม่ แล้วเข้าไปใน container jupyter
docker compose exec jupyter bash

# รัน netcat listener ที่พอร์ต 9999 (พิมพ์ข้อความแล้วกด Enter เพื่อส่งเข้า stream)
nc -lk 9999

2Streaming Word Count

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split

spark = SparkSession.builder \
    .appName("Lab4-Streaming") \
    .master("spark://spark-master:7077") \
    .getOrCreate()

# อ่านจาก socket — สังเกตว่า .readStream แทนที่ .read เท่านั้นที่ต่างจาก batch
lines = spark.readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load()

words = lines.select(explode(split(lines.value, " ")).alias("word"))
word_counts = words.groupBy("word").count()

query = word_counts.writeStream \
    .outputMode("complete") \
    .format("memory") \
    .queryName("word_counts") \
    .start()

# ใน cell ถัดไปของ notebook — เรียกดูผลลัพธ์ที่อัปเดตแล้ว
spark.sql("SELECT * FROM word_counts ORDER BY count DESC").show()

3Windowed Aggregation ตามเวลา

from pyspark.sql.functions import window, current_timestamp

lines_with_ts = spark.readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load() \
    .withColumn("timestamp", current_timestamp())

windowed_counts = lines_with_ts \
    .select(explode(split(lines_with_ts.value, " ")).alias("word"), "timestamp") \
    .groupBy(
        window("timestamp", "30 seconds", "10 seconds"),  # window 30 วิ เลื่อนทุก 10 วิ
        "word"
    ).count()

query2 = windowed_counts.writeStream \
    .outputMode("complete") \
    .format("memory") \
    .queryName("windowed_counts") \
    .start()

# ตรวจสอบผลลัพธ์ที่แบ่งตาม window เวลา
spark.sql("""
    SELECT window.start, window.end, word, count
    FROM windowed_counts
    ORDER BY window.start DESC, count DESC
""").show(truncate=False)

4หยุด Query และทำความสะอาด

query.stop()
query2.stop()
# กด Ctrl+C ใน terminal ที่รัน netcat เพื่อปิด socket server

5บันทึกไว้สำหรับขั้นตอนถัดไป: จาก Socket สู่ Kafka

# จาก socket (Lab นี้)
spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()

# เป็น Kafka (production)
spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "my-topic") \
    .load()

6Event Time กับ Processing Time

Socket Source มีคอลัมน์ timestamp ที่ระบบรับข้อมูล เหมาะสาธิต Processing Time แต่ยังไม่ใช่ Event Time จากต้นทาง สร้าง Stream จากไฟล์ JSON ที่มีเวลาของเหตุการณ์เพื่อทดลอง Late Data

# Terminal สร้างโฟลเดอร์ /home/jovyan/data/events ก่อน
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("event_time", TimestampType(), False),
    StructField("word", StringType(), False)
])

events = spark.readStream.schema(schema).json("/home/jovyan/data/events")

เขียนไฟล์ทีละไฟล์เข้าโฟลเดอร์ โดยให้ Event Time สลับลำดับ สังเกตว่า Window ใช้เวลาใน Record ไม่ใช่เวลาที่ไฟล์มาถึง

7Watermark และ Late Data Experiment

windowed = events.withWatermark("event_time", "2 minutes") \
  .groupBy(window("event_time", "1 minute"), "word") \
  .count()

query = windowed.writeStream \
  .outputMode("append") \
  .format("console") \
  .option("truncate", False) \
  .option("checkpointLocation", "/home/jovyan/data/checkpoints/watermark-lab") \
  .start()

ส่ง Event เวลา 10:00, 10:03 แล้วค่อยส่ง 10:00:30 จดว่า Event สายถูกนับหรือไม่ Watermark ไม่ใช่ Timer ที่รอทุก Event สองนาที แต่ติดตาม Max Event Time แล้วใช้ขอบเขตล้าง State ตาม Operator/Mode

8Output Mode: Append, Update, Complete

รัน Query เดียวกันคนละ Checkpoint/Sink แล้วเปรียบเทียบจำนวนแถวที่พิมพ์ ไม่ใช่ทุก Mode ใช้ได้กับทุก Query ให้บันทึก Error พร้อมอธิบาย Semantic แทนการแก้จนหายโดยไม่รู้สาเหตุ

9Checkpoint และ Recovery

เริ่ม File Stream Query พร้อม Checkpoint เพิ่มไฟล์สองชุด หยุด Query/Kernel แล้วเปิดใหม่ด้วย Code และ Checkpoint เดิม เพิ่มไฟล์ชุดที่สาม ตรวจว่าไฟล์เดิมไม่ถูกประมวลผลซ้ำตาม Progress ที่กู้คืน

query = events.writeStream \
  .format("parquet") \
  .option("path", "/home/jovyan/data/stream-output") \
  .option("checkpointLocation", "/home/jovyan/data/checkpoints/recovery") \
  .start()

บันทึกจำนวน Output ก่อน/หลัง Restart Checkpoint ไม่ได้เก็บข้อมูล Source เอง ถ้า Source ลบข้อมูลเก่าก่อน Recovery สำเร็จ ระบบอาจกู้ไม่ได้ Production จึงต้องประสาน Retention ของ Kafka/Source กับ Recovery Time

10Deduplication ด้วย Event ID

unique = events.withWatermark("event_time", "10 minutes") \
  .dropDuplicates(["event_id"])

unique.writeStream.format("console") \
  .outputMode("append") \
  .option("checkpointLocation", "/home/jovyan/data/checkpoints/dedup") \
  .start()

ส่ง Event ID เดิมสองครั้งโดยค่าอื่นเหมือน/ต่างกัน สังเกตผล Event ID ต้องคงเดิมเมื่อ Producer Retry ถ้า Retry สร้าง UUID ใหม่ Downstream แยก Duplicate ไม่ออก State Dedup ยังต้องมี Watermark/TTL เพื่อไม่จำ ID ตลอดโลก

11State Growth และ Query Progress

import json, time
for _ in range(5):
    time.sleep(5)
    p = query.lastProgress
    if p:
        print(json.dumps(p, indent=2))

หา inputRowsPerSecond, processedRowsPerSecond, Duration, Watermark และ State Operator Rows ถ้า Key ใหม่เพิ่มไม่สิ้นสุดและไม่มีการล้าง State ระบบจะโตแม้อัตรา Input คงที่

12Backpressure จำลอง

เพิ่ม UDF ที่ช้าหรือส่งไฟล์เร็วเกิน Processing เพื่อสร้าง Backlog วัด Input/Processed Rate และ Batch Duration เป้าหมายคือเห็นว่า Broker/Directory ช่วยพัก แต่ไม่แก้สมดุลระยะยาวถ้าอัตราประมวลผลต่ำกว่าอัตราเข้า

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import time

@udf(StringType())
def slow_identity(x):
    time.sleep(0.001)
    return x

slow_stream = events.withColumn("word2", slow_identity("word"))

UDF นี้มีไว้สร้างอาการ ไม่ใช่ Production Solution หลังทดลองให้ลบ แล้วอธิบายวิธีแก้จริง เช่นเพิ่ม Partition ลดงานต่อ Event ใช้ Built-in Function หรือปรับ Sink

13Smart Window Challenge

เปรียบเทียบสามแบบ: เก็บ Event ทั้ง Window, Incremental Sum ที่บวกใหม่/ลบเก่า และ Approximate Summary จากแนวคิด DGIM/Count-Min Sketch ไม่ต้องเขียน DGIM เต็มถ้าเวลาไม่พอ แต่ต้องระบุ State ที่แต่ละแบบเก็บ Error และ Query ที่ตอบได้

14สิ่งที่ต้องส่งและสรุปขั้นตอนถัดไป

• Processing-time และ Event-time Query
• Late Data Timeline พร้อมผล Watermark
• Output Mode Comparison
• Restart/Checkpoint Evidence
• Deduplication Test
• Query Progress และ State Metric
• Smart Window Trade-off Table

สรุปและขั้นตอนถัดไป: Stream ทำให้เวลาและ State กลายเป็นส่วนหนึ่งของคำตอบ Lab 5 จะเปลี่ยนจากความสัมพันธ์ตามเวลาไปเป็นความสัมพันธ์ตาม Edge แล้วตรวจว่า PageRank, Motif และ Component คำนวณอะไร รวมทั้งเหตุใด Graph Schema ผิดแล้ว Algorithm ที่รันถูกก็ยังตอบผิดได้

20วิธีทำแล็บให้ได้มากกว่าคำว่า “รันผ่าน”

แล็บนี้ไม่ได้วัดว่าเราพิมพ์คำสั่งตามตัวอย่างได้ครบหรือไม่ เพราะการรันผ่านอาจเกิดจากความบังเอิญ ค่า Default หรือข้อมูลที่เล็กเกินไปจนมองไม่เห็นปัญหา เป้าหมายคือฝึกตั้งสมมติฐาน เปลี่ยนตัวแปรครั้งละอย่าง เก็บหลักฐาน และอธิบายผลด้วยแนวคิดของระบบกระจาย ถ้าผลไม่ตรงกับที่คาด นั่นไม่ใช่ความล้มเหลว แต่เป็นจุดเริ่มต้นของคำถามที่ดี

1. คาดก่อนรัน: เขียนหนึ่งหรือสองประโยคว่าคิดว่าจะเกิดอะไรและเพราะอะไร
2. กำหนดสิ่งที่คุม: ใช้ข้อมูล โค้ด และทรัพยากรเดิม แล้วเปลี่ยนเฉพาะตัวแปรที่กำลังศึกษา
3. เก็บหลักฐาน: บันทึกเวลา แผนการทำงาน Metric, Log หรือภาพ Spark UI ไม่อาศัยความรู้สึกว่า “เร็วขึ้น”
4. อธิบายกลไก: เชื่อมผลกับ Partition, Shuffle, State, Memory, Network หรือ Scheduling
5. บอกข้อจำกัด: ผลจาก Laptop และข้อมูลจำลองอาจไม่เหมือน Production ให้ระบุสิ่งที่ยังสรุปไม่ได้

21สมุดบันทึกการทดลอง

อย่าจดเฉพาะผลลัพธ์สุดท้าย ควรจด Environment ด้วย เช่น Spark version, Python version, จำนวน Core, Memory, จำนวน Partition, Seed และขนาดข้อมูล เปรียบเหมือนใบเสร็จและสูตรอาหารที่ทำให้คนอื่นย้อนทำจานเดิมได้ หากไม่มีข้อมูลเหล่านี้ ผลที่ดูน่าเชื่อถืออาจทำซ้ำไม่ได้

22ชุดการทดลองหลัก: Structured Streaming

23บันไดแก้ปัญหาเมื่อผลไม่เป็นอย่างที่คิด

เวลางานล้ม อย่าเริ่มด้วยการลบทุกอย่างแล้วติดตั้งใหม่ เพราะเราอาจทำลายหลักฐานที่บอกสาเหตุ ให้ไล่จากชั้นนอกเข้าสู่ชั้นในเหมือนช่างไฟที่ตรวจตั้งแต่ปลั๊ก สายไฟ ฟิวส์ ไปจนถึงเครื่องใช้ไฟฟ้า

1. ยืนยันอาการ: คัดลอก Error แรก ไม่ใช่เฉพาะบรรทัดสุดท้าย และบันทึกคำสั่งที่ทำให้เกิดซ้ำ
2. ลดปัญหา: ใช้ข้อมูลเล็กที่สุดและโค้ดสั้นที่สุดที่ยังทำให้ Error เกิด
3. ตรวจสัญญาข้อมูล: ดู Schema, Type, Null, Key และ Path ก่อนสงสัย Cluster
4. ตรวจ Plan และ UI: ดูว่างานเริ่มจริงหรือค้างอยู่ก่อนสร้าง Job; แยกปัญหา Driver, Executor และ Storage
5. ตรวจทรัพยากร: ดู Memory, Disk, Port, Permission และ Container status
6. เปลี่ยนทีละจุด: เมื่อแก้แล้ว ให้ย้อนอธิบายว่าการเปลี่ยนนั้นจัดการสาเหตุใด

24คำถามชวนคิดหลังทำแล็บ

1. ผลที่เห็นเกิดจาก Algorithm หรือเกิดจากการจัดวางข้อมูลและข้อจำกัดของเครื่อง?
2. ถ้าข้อมูลใหญ่ขึ้นหนึ่งร้อยเท่า ขั้นตอนไหนจะพังก่อน และเรามีหลักฐานอะไร?
3. ถ้าเครื่องหนึ่งหายไประหว่างงาน คำตอบจะยังถูกต้องหรือเพียงแค่งานยังรันต่อได้?
4. ค่า Default ใดช่วยเราอยู่ และค่าใดอาจกลายเป็นกับดักเมื่อขึ้น Production?
5. ถ้าต้องอธิบายผลให้เพื่อนที่ไม่ได้เห็นหน้าจอ เรามีหลักฐานครบพอหรือยัง?

25สิ่งที่ต้องส่ง

• Notebook หรือ Source code ที่รันจากต้นจนจบได้ โดยไม่พึ่งลำดับคำสั่งที่ซ่อนอยู่
• README ระบุวิธีเริ่มระบบ Version, Dependency และคำสั่งที่ใช้
• สมุดบันทึกอย่างน้อยสามการทดลอง มีสมมติฐาน ตัวแปร หลักฐาน และข้อสรุป
• ภาพหรือ Export จาก Spark UI/Query Plan ที่วงส่วนสำคัญและมีคำอธิบาย
• กรณีผิดพลาดอย่างน้อยหนึ่งกรณี พร้อมการวิเคราะห์ ไม่ตัดทิ้งเพราะ “ทำไม่สำเร็จ”
• ข้อจำกัดของผลและสิ่งที่จะทดลองต่อ หากมีข้อมูลหรือเครื่องมากขึ้น

26เกณฑ์ตรวจงานแบบสั้น

27สรุปและขั้นตอนถัดไป

หัวใจของแล็บนี้ไม่ใช่จำคำสั่ง แต่คือการมองเห็นว่า Structured Streaming ตัดสินใจอย่างไรเมื่อข้อมูล งาน และทรัพยากรถูกกระจายออกจากกัน เมื่อทำเสร็จแล้วควรตอบได้สามเรื่อง: ระบบทำอะไร หลักฐานใดแสดงว่ามันทำเช่นนั้น และเงื่อนไขใดทำให้พฤติกรรมเปลี่ยน

ก่อนเปิดแล็บถัดไป ลองย่อสิ่งที่เรียนรู้ให้เหลือหนึ่งภาพ หนึ่งตาราง และสามประโยค หากย่อไม่ได้ อาจยังมีส่วนที่เราเห็นผลแต่ยังไม่เข้าใจกลไก ให้กลับไปเลือกการทดลองที่เล็กลง แล้วค่อยต่อภาพกลับขึ้นมาใหม่

28สถานการณ์จำลอง: จากข้อมูลก้อนเล็กไปสู่งานที่เริ่มมีปัญหา

ให้เริ่มด้วยข้อมูลก้อนเล็กที่ตรวจคำตอบด้วยมือได้ จากนั้นขยายเป็นสามระดับ โดยไม่เปลี่ยนความหมายของโจทย์ ระดับแรกใช้ตรวจความถูกต้อง ระดับที่สองใช้เปิดให้เห็น Parallelism และระดับที่สามใช้เปิดให้เห็นคอขวด วิธีนี้ช่วยแยกคำถามสองข้อที่มักปนกัน คือ “คำตอบถูกหรือไม่” กับ “ระบบรองรับขนาดงานได้หรือไม่” ระบบที่เร็วแต่ตอบผิดไม่มีประโยชน์ และระบบที่ตอบถูกเฉพาะข้อมูลสิบแถวก็ยังไม่ใช่คำตอบของงานขนาดใหญ่

29อ่าน Spark UI ให้เป็นเรื่องราว ไม่ใช่เพียงหน้าจอสีสวย

เริ่มจาก Job ว่าถูกสร้างเพราะ Action ใด แล้วลงไปที่ Stage เพื่อดูว่า Shuffle แบ่งเส้นทางตรงไหน จากนั้นจึงดู Task ว่ากระจายตัวสม่ำเสมอหรือมีตัวใดลากยาว สุดท้ายดู Executor ว่างานไปอยู่เครื่องใด ใช้ Memory เท่าไร และมี Failure หรือ Retry หรือไม่ ลำดับนี้เหมือนอ่านแผนที่จากประเทศ จังหวัด ถนน แล้วจึงถึงบ้าน หากกระโดดไปดู Task หนึ่งตัวทันที เราอาจเห็นอาการแต่ไม่เห็นบริบท

การจับภาพ UI ควรใส่ลูกศรหรือคำอธิบายว่ากำลังใช้ภาพนั้นสนับสนุนข้อสรุปใด ภาพทั้งหน้าที่ไม่มีคำอธิบายเหมือนแนบผลตรวจสุขภาพโดยไม่บอกว่าค่าไหนผิดปกติ หลักฐานที่ดีไม่จำเป็นต้องเยอะ แต่ต้องเชื่อมกับคำถามได้ตรงจุด

30แยกความถูกต้อง ความเร็ว และความทนทานออกจากกัน

สามเรื่องนี้สัมพันธ์กันแต่ทดแทนกันไม่ได้ งานอาจเร็วขึ้นเพราะเผลอทิ้งข้อมูลบางส่วน งานอาจทนต่อ Worker หายแต่สร้างข้อมูลซ้ำ หรือคำตอบอาจถูกแต่ใช้ทรัพยากรมากเกินจำเป็น ทุกครั้งที่ปรับระบบจึงควรตรวจทั้งสามแกน ไม่ประกาศชัยชนะจากเวลาเพียงตัวเดียว

1. ความถูกต้อง: สร้างข้อมูลจิ๋วที่รู้คำตอบล่วงหน้า ใส่ Null, Duplicate, Key ที่ไม่มีคู่ และค่าขอบเขต แล้วตรวจผลทีละกรณี
2. ความเร็ว: วัดหลายรอบ แยก Warm-up ออกจากรอบจริง ระบุขนาดข้อมูลและทรัพยากร และรายงานทั้งค่ากลางกับช่วง ไม่เลือกเฉพาะรอบที่ดีที่สุด
3. ความทนทาน: ทำให้ส่วนหนึ่งล้มอย่างตั้งใจ ตรวจว่างาน Retry, Resume หรือคำนวณใหม่จาก Lineage และตรวจว่าคำตอบหลังการกู้ไม่ซ้ำหรือขาด
4. ต้นทุน: เวลาเร็วขึ้นอาจแลกกับ Memory, Storage, Network หรือความซับซ้อนในการดูแล ให้บันทึกสิ่งที่จ่ายเพิ่มเสมอ

31จาก Error Message ไปสู่คำอธิบาย

ให้เลือก Error จริงหนึ่งครั้งจากแล็บแล้วเขียน Postmortem สั้น ๆ โดยไม่กล่าวโทษผู้ใช้หรือเครื่องมือ เริ่มจากผลกระทบ ตามด้วยลำดับเหตุการณ์ สาเหตุใกล้ตัว สาเหตุราก วิธีแก้เฉพาะหน้า และวิธีป้องกันซ้ำ ตัวอย่างเช่น “Job ล้มเพราะ Out of Memory” ยังเป็นเพียงอาการ สาเหตุรากอาจเป็นการ collect ข้อมูลทั้งหมดมายัง Driver, Broadcast ตารางใหญ่เกินไป หรือ Partition หนึ่งก้อนใหญ่ผิดปกติ ซึ่งมีวิธีแก้คนละแบบ

จุดสำคัญคือแยก “แก้ให้งานผ่านครั้งนี้” ออกจาก “แก้ไม่ให้เกิดซ้ำ” การเพิ่ม Memory อาจช่วยงานรอบนี้ แต่ถ้าข้อมูลโตต่อเนื่อง ปัญหาจะกลับมา การเปลี่ยนรูปแบบ Aggregation, ลดข้อมูลก่อน Shuffle หรือเลิกดึงทุกอย่างเข้า Driver อาจแตะสาเหตุได้ตรงกว่า

32แบบฝึกสอนกลับ: อธิบายให้เพื่อนฟังในห้านาที

จับคู่กับเพื่อนแล้วผลัดกันอธิบาย Structured Streaming โดยห้ามเริ่มจากชื่อ Class หรือ API ให้เริ่มจากปัญหาที่ระบบกำลังแก้ ใช้ภาพหนึ่งภาพและตัวอย่างข้อมูลไม่เกินสิบแถว ผู้ฟังมีหน้าที่ถามว่า “รู้ได้อย่างไร” และ “ถ้าข้อมูลใหญ่ขึ้นจะเกิดอะไร” วิธีนี้ช่วยเปิดจุดที่เราจำคำสั่งได้แต่ยังเชื่อมเหตุผลไม่ครบ

• นาทีที่ 1: ปัญหาคืออะไร และเหตุใดเครื่องเดียวหรือวิธีตรงไปตรงมาจึงเริ่มไม่พอ
• นาทีที่ 2: ข้อมูลและงานถูกแบ่งอย่างไร ส่วนใดอยู่ที่ Driver และส่วนใดอยู่ที่ Executor
• นาทีที่ 3: จุดใดต้องสื่อสารข้ามเครื่อง และต้นทุนเกิดตรงไหน
• นาทีที่ 4: แสดงหลักฐานหนึ่งชิ้นจากแล็บแล้วตีความ
• นาทีที่ 5: บอกข้อจำกัดหนึ่งข้อและการทดลองถัดไปหนึ่งอย่าง