Streaming แบบเรียลไทม์

1Data Stream Management System (DSMS) และ Complex Event Processing (CEP)

Stream เป็นข้อมูลที่ไม่มีขอบเขต แต่คำถามต้องมีขอบเขต

ใน Batch เราอ่านไฟล์ตั้งแต่ต้นจนจบแล้วสรุปผลได้ แต่ Stream ไม่มี EOF ที่มีความหมายทางธุรกิจ การถามว่า “ยอดขายทั้งหมดของ Stream เท่าไร” จึงเป็นคำถามที่ไม่จบ หรือกลายเป็นค่าที่เปลี่ยนตลอดเวลา วิธีทำให้ตอบได้คือกำหนดขอบเขต เช่นยอดขายตั้งแต่เปิดร้าน ยอดขาย 15 นาทีล่าสุด หรือยอดขายต่อ Session ของลูกค้า

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

DSMS, Stream Analytics และ CEP ซ้อนกัน แต่ไม่ใช่เรื่องเดียวกัน

ระบบหนึ่งอาจมีทั้งสี่แบบ เช่นแพลตฟอร์มธนาคารนับธุรกรรมต่อบัญชี ตรวจ pattern ฉ้อโกง ใช้โมเดลให้คะแนน และสั่งระงับบัตร ความต่างอยู่ที่คำถามและ state ไม่ใช่ชื่อผลิตภัณฑ์ที่เลือกใช้

2เวลาสามแบบที่ต้องแยกให้ออก

ความผิดพลาดที่พบบ่อยที่สุดใน Streaming คือใช้คำว่า “เวลา” เหมือนมีความหมายเดียว ในความจริงเหตุการณ์หนึ่งมีเวลาได้หลายค่า และแต่ละค่าตอบคนละคำถาม

ทำไม Processing Time จึงง่ายแต่เปลี่ยนตามสภาพระบบ

การแบ่งหน้าต่างตามเวลาที่ระบบประมวลผลง่าย เพราะนาฬิกาของระบบพร้อมใช้และไม่ต้องจัดการเหตุการณ์มาช้า แต่ถ้าคลัสเตอร์ติด Backlog เหตุการณ์ที่เกิดในนาทีเดียวกันอาจถูกแยกคนละหน้าต่าง เพียงเพราะบางข้อความรอคิว การ Restart หรือเพิ่มทรัพยากรยังอาจทำให้ผลต่างกันทั้งที่ input เดิม

Event Time สะท้อนธุรกิจมากกว่า แต่ต้องยอมรับความยุ่งยาก

Event Time ช่วยให้ยอดของช่วง 10:00–10:05 หมายถึงเหตุการณ์ที่เกิดจริงในช่วงนั้น ไม่ว่าข้อความจะมาถึงเมื่อใด แต่ระบบต้องเก็บ state รอข้อมูลที่ยังมาไม่ถึง และต้องตัดสินใจว่า “ช้าเกินไป” คือกี่นาที ถ้ารอตลอดไปคำตอบสมบูรณ์ขึ้น แต่ state ก็ไม่ถูกล้างและ dashboard ไม่เคยนิ่ง

Out-of-Order ไม่ได้แปลว่าข้อมูลผิด

ข้อความอาจสลับลำดับเพราะ mobile device offline, network retry, partition ต่างกัน หรือ producer หลายตัวมีนาฬิกาไม่ตรงกัน ระบบจึงไม่ควรตีความว่าข้อความมาทีหลังต้องเกิดทีหลังเสมอ ลำดับที่ต้องการอาจเป็นลำดับต่อ device, ต่อ account หรือทั่วทั้งระบบ ซึ่งมีต้นทุนและความเป็นไปได้ต่างกัน

3Window: วิธีแบ่งสิ่งที่ไม่มีวันจบให้ตอบได้

Window เป็นทั้งขอบเขตเชิงคำนวณและขอบเขตเชิงความหมาย เราต้องเลือกจากคำถาม ไม่ใช่เลือกเพราะ API เขียนง่ายที่สุด

Tumbling Window: ง่ายแต่มีปัญหาที่ขอบ

ถ้าเหตุการณ์สำคัญสองรายการเกิดเวลา 10:04:59 และ 10:05:01 มันจะอยู่คนละหน้าต่าง 5 นาที แม้ห่างกันเพียงสองวินาที สำหรับรายงานตามช่วงเวลานี่อาจถูกต้อง แต่สำหรับตรวจ burst หรือ pattern อาจทำให้พลาด Sliding Window ช่วยลดปัญหาขอบโดยให้หน้าต่างทับกัน แลกกับการคำนวณมากขึ้น

Sliding Window: ขนาดกับระยะเลื่อนเป็นคนละค่า

Window 30 นาทีที่เลื่อนทุก 5 นาที หมายความว่า event หนึ่งอาจมีส่วนใน 6 windows ไม่ใช่เพียงหนึ่ง ยิ่ง window ยาวและ slide สั้น จำนวนกลุ่มที่ต้องอัปเดตยิ่งมาก การตั้งค่าให้ dashboard “สด” ทุกวินาทีอาจสร้างงานซ้ำจำนวนมหาศาลทั้งที่ผู้ใช้ดูผลทุกหนึ่งนาที

Session Window: ให้ข้อมูลบอกขอบเขต

Session ไม่มีเวลาเริ่มจบคงที่ ถ้าผู้ใช้มี event ใหม่ก่อน inactivity gap หมด session จะยาวออกไป เหมือนบทสนทนาที่ถือว่ายังต่อเนื่องตราบใดที่ยังไม่มีช่วงเงียบนานเกินกำหนด เหมาะกับพฤติกรรมผู้ใช้ เครื่องจักร หรือธุรกรรมที่มีช่วงกิจกรรมเป็นชุด

import org.apache.spark.sql.functions._

val events = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "events")
  .load()
  .select(from_json($"value".cast("string"), eventSchema).as("e"))
  .select("e.*")

val counts = events
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window($"event_time", "30 minutes", "5 minutes"),
    $"event_type"
  )
  .count()

โค้ดข้างต้นมีสามระยะที่ต้องไม่สับสน: window ยาว 30 นาที, เลื่อนทุก 5 นาที และ watermark ยอมรับความล่าช้าตามเกณฑ์ 10 นาที แต่ watermark ไม่ได้แปลว่า “รอทุก event 10 นาทีแล้วค่อยแสดง” มันเป็นกลไกติดตามความคืบหน้าของ Event Time และช่วยกำหนดว่า state เก่าพอจะล้างได้เมื่อใด

4Spark Streaming และ Micro-batch: รากทางประวัติศาสตร์

// DStream: ลำดับของ RDD ที่แทน stream ข้อมูล
val tweets = ssc.twitterStream()

// transformation: แปลง DStream หนึ่งเป็นอีกอันหนึ่ง
val hashTags = tweets.flatMap(status => getTags(status))

// windowed operation: นับ hashtag ในช่วง 1 นาทีล่าสุด อัปเดตทุก 5 วินาที
val tagCounts = hashTags.window(Minutes(1), Seconds(5)).countByValue()

// output operation: ส่งผลลัพธ์ไปเก็บที่ HDFS
hashTags.saveAsHadoopFiles("hdfs://...")

// วิธีไร้เดียงสา: คำนวณ count ใหม่ทั้ง window ทุกครั้ง (ช้า)
val tagCounts = hashtags.countByValueAndWindow(Minutes(10), Seconds(1))

// วิธีฉลาด: "บวก" ค่าจาก batch ใหม่ และ "ลบ" ค่าจาก batch ที่หลุดจาก window
val tagCounts = hashtags.reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(1))

Micro-batch ไม่ใช่ Batch ธรรมดาที่ตั้งเวลาให้ถี่ขึ้น

DStream แบ่ง input เป็น RDD ตามช่วงเวลา แต่ระบบยังต้องดูแล receiver, checkpoint, scheduling และการกู้คืนต่อเนื่อง ความคิดสำคัญคือเอากลไก fault tolerance ของ RDD มาใช้กับข้อมูลไหล วิธีนี้ให้ throughput สูงและรวมกับ Spark ecosystem ได้ดี แต่ latency ถูกผูกกับ batch interval และเวลาที่แต่ละ batch ใช้ประมวลผล

Batch Interval คือสัญญาระหว่าง Latency กับ Overhead

ถ้าตั้ง batch ทุก 100 มิลลิวินาที ระบบต้องสร้างและจัดตารางงานถี่มาก overhead อาจกินเวลาส่วนใหญ่ แต่ถ้าตั้งทุก 1 นาที ผลตอบสนองช้าและ batch หนึ่งอาจใหญ่เกินทรัพยากร หลักพื้นฐานคือเวลาเฉลี่ยและ tail processing time ของ batch ควรต่ำกว่า batch interval อย่างมีระยะเผื่อ มิฉะนั้น batch ใหม่จะเข้ามาเร็วกว่าที่ batch เก่าถูกระบาย เกิด backlog สะสม

Inverse Reduce ใช้ได้เมื่อ operation ย้อนกลับได้

Sliding sum สามารถบวกค่าที่เข้ามาใหม่และลบค่าที่หมดอายุ แต่ไม่ใช่ทุก aggregate มี inverse ที่ง่าย เช่นค่า MAX ถ้าค่าสูงสุดหลุดจาก window เราไม่รู้ค่าสูงสุดตัวใหม่จากค่า MAX เดิมเพียงค่าเดียว ต้องเก็บโครงสร้างเพิ่มหรือคำนวณใหม่บางส่วน ส่วน median และ quantile ยิ่งต้องใช้ state ซับซ้อนหรือ approximation

5Watermark และ Late Data: จะรออดีตนานแค่ไหน

เมื่อใช้ Event Time ระบบไม่รู้ว่าข้อมูลของ window หนึ่งมาครบหรือยัง เพราะข้อความที่เก่ากว่าอาจกำลังติดอยู่ในเครือข่าย Watermark เป็นกลไกบอกความคืบหน้าของเวลาเชิงเหตุการณ์ เพื่อช่วยตัดสินใจว่า state ของอดีตช่วงใดเก่าพอจะล้างได้

Watermark ไม่ใช่กำแพงตรงเวลาที่รับประกันทุกกรณี

ผู้เรียนมักตีความว่าเหตุการณ์ที่ช้ากว่า 10 นาที “ถูกทิ้งแน่นอน” และเหตุการณ์ที่ช้าไม่เกิน 10 นาที “รับแน่นอน” ในทางปฏิบัติ semantics ขึ้นกับ operator, output mode และความคืบหน้าของ query เอกสาร Spark อธิบาย guarantee ในด้านที่ข้อมูลช้าภายในเกณฑ์จะไม่ถูกทิ้ง ส่วนข้อมูลที่ช้ากว่าเกณฑ์อาจถูกนำมารวม หรืออาจถูกทิ้งเมื่อ state ถูกล้างแล้วก็ได้ เราจึงไม่ควรใช้ watermark เป็นเครื่องตรวจเวลาสายแบบ deterministic โดยไม่ดู semantics ของ operation

เกณฑ์ Watermark ควรมาจาก Distribution ของความล่าช้า

การเลือก 10 นาทีเพราะเป็นเลขกลมไม่ใช่การออกแบบที่ดี เราควรวัด ingestion_time - event_time ตามแหล่งข้อมูล ดู percentile และดูเหตุการณ์ผิดปกติ เช่น mobile offline หลายชั่วโมง จากนั้นค่อยตัดสินใจว่าอยากครอบคลุมข้อมูลกี่เปอร์เซ็นต์ และรับ state/latency ได้เท่าไร

Late Data มีทางออกมากกว่าทิ้งหรือรอ

• Update ผลย้อนหลัง: เหมาะเมื่อ sink รองรับ update/upsert และผู้ใช้ยอมรับตัวเลขที่เปลี่ยนได้
• ส่งเข้า Dead-letter หรือ Late-event Table: เก็บเหตุการณ์สายเกินเกณฑ์ไว้ตรวจหรือชดเชยภายหลัง
• Reconciliation แบบ Batch: Streaming ให้คำตอบเร็ว แล้ว Batch รายวันคำนวณผลที่ครบกว่าเพื่อแก้ยอด
• แยก SLA ตามแหล่ง: อุปกรณ์ในโรงงานอาจมาช้าไม่เกิน 1 นาที แต่มือถืออาจ offline หลายชั่วโมง ไม่จำเป็นต้องใช้เกณฑ์เดียวกัน

6Stateful Streaming: สิ่งที่ระบบต้องจำระหว่างเหตุการณ์

Operation เช่น select หรือ filter ประมวลผลแต่ละ event ได้โดยไม่จำอดีต เรียกว่า stateless แต่การนับต่อ key, window aggregation, deduplication, stream-stream join, session และ CEP ต้องเก็บ state ข้ามเหตุการณ์ ความยากของ Streaming ส่วนใหญ่จึงซ่อนอยู่ในคำถามว่า “ต้องจำอะไร และลืมเมื่อใด”

State โตจาก Cardinality × ช่วงเวลาที่เก็บ

ถ้ามีผู้ใช้ 100 ล้านรายและเก็บ state หนึ่งก้อนต่อผู้ใช้ แม้ก้อนละไม่กี่ร้อย byte ก็กลายเป็นหลายสิบ GB ยังไม่รวม index, serialization และ checkpoint ถ้า key ใหม่เพิ่มต่อเนื่องโดยไม่มี TTL state จะโตอย่างไม่มีขอบเขต แม้อัตรา event ต่อวินาทีคงที่

Hot Key ใน Streaming

Key UNKNOWN หรือบัญชียอดนิยมอาจรับ event มากกว่าค่าเฉลี่ยหลายล้านเท่า ทำให้ partition หนึ่งแบก state และงานหนักกว่าส่วนอื่น การเพิ่ม executor ไม่ช่วยถ้า hot key ยังต้องอยู่ task เดียวเพื่อรักษาความถูกต้อง วิธีแก้อาจเป็นแยก key พิเศษ, ทำ local aggregation, salting แล้วรวมสองชั้น หรือเปลี่ยนนิยามผลลัพธ์

Custom Stateful Processing

Built-in window และ aggregation เหมาะกับโจทย์มาตรฐาน แต่บางงานต้องเก็บ state หลายค่า ตั้ง timer หรือจัดการ lifecycle เฉพาะ ใน Spark 4.x มี transformWithState เป็น API รุ่นใหม่สำหรับ arbitrary stateful processing ต่อ key แนวคิดสำคัญกว่าชื่อ API คือ state ต้องมี schema, การหมดอายุ และเส้นทางกู้คืนที่ชัด ไม่ใช่ HashMap ที่ปล่อยให้โตไปเรื่อย ๆ

7Smart Window Counting: จำให้น้อย แต่ตอบได้มาก

Incremental Window ยังถือว่าเราจำค่าที่ต้องนำออกจากหน้าต่างได้ แต่ถ้า stream เร็วมากหรือหน้าต่างยาวมาก การเก็บทุกเหตุการณ์อาจแพงเกินไป เราจึงยอมให้คำตอบคลาดเคลื่อนเล็กน้อย แลกกับหน่วยความจำที่ลดลงมาก

เริ่มจากโจทย์ที่ง่ายแต่เก็บตรง ๆ ไม่ไหว

สมมุติมี stream ของ bit โดย 1 หมายถึงเกิดเหตุการณ์ที่สนใจ และ 0 หมายถึงไม่เกิด เราต้องตอบว่าใน k ตำแหน่งล่าสุดมีเลข 1 กี่ตัว โดย k ≤ N และ window ใหญ่สุดยาว N วิธีตรงไปตรงมาคือเก็บ bit ทั้ง N ตัวหรือเก็บ timestamp ของทุกเลข 1 แต่ถ้า N ใหญ่มากหรือมี stream หลายล้าน key หน่วยความจำจะกลายเป็นข้อจำกัดหลัก

DGIM เปลี่ยนคำถามจาก “จำทุกเหตุการณ์อย่างไร” เป็น “ต้องจำโครงสร้างอะไรจึงประมาณคำตอบได้” จุดเปลี่ยนนี้เป็นแก่นของ Streaming Algorithm เราไม่ได้ย่อข้อมูลหลังคำนวณเสร็จ แต่สร้าง summary ที่อัปเดตได้ตั้งแต่ข้อมูลไหลเข้ามา

Bucket ใน DGIM เก็บอะไร

Bucket หนึ่งเก็บอย่างน้อยสองอย่าง คือ ขนาด ซึ่งเป็นจำนวนเลข 1 ใน bucket และ timestamp ของเลข 1 ตัวล่าสุดใน bucket ขนาดเป็นกำลังของสอง ได้แก่ 1, 2, 4, 8, ... และรักษากฎว่าแต่ละขนาดมี bucket ได้ไม่เกินจำนวนที่กำหนด ในเวอร์ชันพื้นฐานมักไม่เกินสอง bucket ต่อขนาด

เมื่อเลข 1 ใหม่เข้ามา Algorithm ทำอะไร

1. สร้าง bucket ขนาด 1 ที่ timestamp ปัจจุบัน
2. ถ้ามี bucket ขนาด 1 เกินสองก้อน รวมสองก้อนที่เก่าที่สุดเป็น bucket ขนาด 2
3. ถ้าการรวมทำให้ขนาด 2 เกินสองก้อน รวมสองก้อนเก่าที่สุดเป็นขนาด 4
4. ทำต่อเหมือนการทดเลข จนทุกขนาดกลับมาตรงกฎ
5. ทิ้ง bucket ที่หมดอายุจาก window N

ตัวอย่างการอัปเดต Bucket

ตารางนี้ย่อเฉพาะขนาดเพื่อเห็นรูปแบบ ใน implementation จริงต้องเก็บ timestamp ของขอบ bucket ด้วย เพราะเลข 0 ทำให้ระยะตำแหน่งไม่เท่ากับจำนวนเลข 1

ตอบ Query ใน k ตำแหน่งล่าสุดอย่างไร

เราไล่ bucket จากใหม่ไปเก่า บวกขนาดเต็มของ bucket ที่อยู่ในช่วง k ทั้งหมด เมื่อถึง bucket เก่าสุดที่คาบขอบ window เราไม่รู้ว่าเลข 1 ใน bucket นั้นอยู่ด้านในกี่ตัว จึงประมาณด้วยครึ่งหนึ่งของ bucket นั้น ส่วน bucket ที่เก่ากว่านั้นไม่นับ

estimate = 0
for bucket from newest to oldest:
    if bucket อยู่ภายใน k ทั้งก้อน:
        estimate += bucket.size
    else if bucket คาบขอบของ k:
        estimate += bucket.size / 2
        break

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

พื้นที่จัดเก็บลดลงอย่างไร

มี bucket size เพียงประมาณ log₂N ระดับ แต่ละระดับมีจำนวน bucket คงที่ และแต่ละ bucket ต้องเก็บ size กับ timestamp ที่ใช้ประมาณ log N bit จึงใช้พื้นที่ระดับ O(log²N) bit ในรูปแบบพื้นฐาน แทน O(N) bit ของการเก็บหน้าต่างตรง ๆ ความต่างมหาศาลเมื่อ N ใหญ่และมีหลาย key

ปรับความแม่นยำได้หรือไม่

ได้ แนวคิดทั่วไปคือยอมให้มี bucket ต่อขนาดมากกว่าสองก้อนก่อนรวม เมื่อ bucket เล็กลงเมื่อเทียบกับผลรวม ความคลาดเคลื่อนสัมพัทธ์จากก้อนที่คาบขอบก็ลดลง แต่จำนวน bucket และหน่วยความจำเพิ่มขึ้น นี่เป็น trade-off ที่ตรงไปตรงมา: อยากให้ความผิดพลาดเล็กลง ต้องจ่าย state มากขึ้น

DGIM ไม่ใช่ Window Aggregation ทั่วไปทุกชนิด

DGIM แบบพื้นฐานออกแบบเพื่อประมาณจำนวนเลข 1 ใน suffix ของ binary stream เราประยุกต์กับ integer ขนาดจำกัดผ่านการแยก bit หรือขยายแนวคิดไปยัง summary แบบอื่นได้ แต่ไม่ควรกล่าวว่า DGIM แก้ average, distinct count หรือ top-k โดยตรงทุกอย่าง Algorithm แต่ละแบบรักษาข้อมูลไม่เหมือนกัน เพราะคำถามที่ต้องตอบต่างกัน

เลือก Summary ให้ตรงคำถาม

Mergeability: คุณสมบัติสำคัญในระบบกระจาย

Summary ที่ดีในระบบกระจายควรรวมผลจากหลาย partition ได้ โดยไม่ต้องดึง raw event ทั้งหมดกลับมา เช่น Count-Min Sketch และ HyperLogLog สามารถ merge summary จากหลายเครื่องได้ภายใต้เงื่อนไขที่ตรงกัน คุณสมบัตินี้เชื่อมกับ Monoidification จากบท Functional Programming: ถ้าสถานะมี operation รวมที่ associative เราสามารถ aggregate เป็นชั้น ๆ ลดข้อมูลก่อนข้ามเครือข่าย และกู้คืนงานได้ง่ายขึ้น

Exact หรือ Approximate ไม่ใช่คำถามทางเทคนิคล้วน ๆ

ยอดเงินในบัญชีอาจต้อง exact และตรวจสอบย้อนหลังได้ แต่จำนวนผู้ชมโดยประมาณบน dashboard อาจยอมคลาดเคลื่อน 1% เพื่อออกเร็วและประหยัดทรัพยากร การเลือกต้องระบุ error budget, latency budget, cost budget และผลกระทบเมื่อผิด ไม่ใช่สรุปว่า approximate “ด้อยกว่า” เพราะบางครั้งคำตอบที่แม่น 100% แต่สายไปหนึ่งวันไม่มีประโยชน์

8กรณีศึกษา: Kafka + Spark Streaming + Elasticsearch

ขยาย 300 มิติอย่างไรไม่ให้เกิด Explosion

ถ้า event หนึ่งถูกแตกเป็น aggregate record หลายร้อยชุดอย่างตรงไปตรงมา จำนวน record ระหว่างทางอาจสูงกว่า input หลายร้อยเท่า วิธีแก้คือรวมภายใน partition ก่อน ส่งเฉพาะ partial aggregate ตาม key ที่ต้องการ และแยก query ตาม SLA แทนการคำนวณทุก dimension ทุก granularity ในเส้นเดียว

Kafka ทำหน้าที่เป็น Log ไม่ใช่เพียงท่อส่งข้อความ

Kafka เก็บ record ตาม partition พร้อม offset ทำให้ consumer อ่านต่อหรือ replay ช่วงเดิมได้ Producer และ consumer จึงไม่ต้องทำงานด้วยความเร็วเท่ากันทุกขณะ อย่างไรก็ตาม Kafka ไม่ได้ทำให้ downstream ถูกต้องเอง เราต้องกำหนด key เพื่อรักษาลำดับภายใน partition, retention ให้ยาวพอสำหรับ recovery และตรวจว่า consumer lag โตเกิน SLA หรือไม่

Elasticsearch เหมาะกับการค้น แต่ต้องออกแบบการเขียนซ้ำ

ผล aggregate ที่เขียนไป sink อาจถูก retry หลัง failure ถ้าใช้ document ID แบบสุ่ม การ retry จะสร้างเอกสารซ้ำ แต่ถ้าสร้าง ID แบบ deterministic จาก window + dimensions การเขียนเดิมสามารถ upsert ทับได้ การทำ sink ให้ idempotent จึงเป็นส่วนหนึ่งของความถูกต้อง ไม่ใช่ optimization เสริม

9สถาปัตยกรรม Streaming: Source → Compute → Sink

ระบบ Streaming จริงไม่ได้มีเพียง Spark job หนึ่งตัว แต่เป็นสายตั้งแต่ producer, message log, stream processor, state/checkpoint ไปจน sink และผู้ใช้ปลายทาง ปัญหาที่เกิดในส่วนหนึ่งส่งผลต่อส่วนอื่น เช่น sink ช้าทำให้ processor ช้า consumer lag เพิ่ม และ retention ของ source อาจหมดก่อนกู้ระบบเสร็จ

Partition Key กำหนดทั้งลำดับและการกระจายงาน

ถ้า key ด้วย account_id event ของบัญชีเดียวกันมักไป partition เดียว จึงรักษาลำดับต่อบัญชีและทำ stateful processing ได้ง่าย แต่บัญชียอดนิยมอาจกลายเป็น hot partition ถ้าใช้ random key กระจายดีขึ้นแต่ลำดับและ state ต่อบัญชีซับซ้อนขึ้น Key จึงเป็นการตัดสินใจเชิง semantic และ performance พร้อมกัน

Backpressure: เมื่อปลายทางกินช้ากว่าต้นทางผลิต

ระบบที่รับ 100,000 event/s แต่ประมวลผลได้ 80,000 event/s จะสะสมหนี้ 20,000 event ทุกวินาที ต่อให้ยังไม่ล้ม Latency จะโตและ retention ถูกกินเรื่อย ๆ ทางแก้มีทั้งเพิ่มทรัพยากร ลดงานต่อ event ปรับ partition จำกัด input rate ลดความถี่ output หรือยอมลดรายละเอียดของคำตอบ

Capacity Planning จากอัตราสูงสุด ไม่ใช่ค่าเฉลี่ยอย่างเดียว

Traffic อาจพุ่งช่วงแคมเปญหรือเหตุการณ์ฉุกเฉิน เราควรดู peak rate, burst duration, record size, serialization, amplification จาก Join/Aggregate และเวลาที่ sink ชะลอ ถ้าระบบรองรับค่าเฉลี่ยได้พอดี จะไม่มีพื้นที่เหลือสำหรับ failure หรือ replay

10ความถูกต้องเมื่อข้อความซ้ำและระบบล้ม

คำว่า Exactly-once มักถูกใช้กว้างเกินจริง เราต้องถามว่า exactly-once ที่ส่วนใด: การอ่าน offset, การอัปเดต state, หรือผลข้างเคียงที่ sink ภายนอก ระบบอาจประมวลผล state อย่างถูกต้อง แต่ส่งอีเมลซ้ำสองฉบับหลัง retry ได้

Checkpoint จำอะไร

Structured Streaming ใช้ checkpoint เก็บ progress เช่น offset ที่ประมวลผลแล้ว รวมถึง metadata และ state ของ stateful operator ตาม query การ Restart ด้วย checkpoint เดิมช่วยให้ query อ่านต่อและกู้ state แต่ checkpoint ผูกกับโครงสร้าง query หลายส่วน การแก้ query บางชนิดแล้วใช้ checkpoint เดิมอาจไม่รองรับ จึงต้องมีแผน migration ไม่ใช่เปลี่ยนโค้ดแล้วหวังว่าจะอ่านต่อได้เสมอ

Idempotency: ทำซ้ำแล้วผลเหมือนเดิม

ถ้า sink รับ key เดิมแล้ว upsert ค่าเดิม การ retry ไม่ทำให้ผลสะสมซ้ำ เช่นเขียนยอดของ window ด้วย key (window_start, customer_id) แต่คำสั่ง “เพิ่มยอดอีก 100” ไม่ idempotent ถ้าถูกเรียกสองครั้ง เราอาจเปลี่ยนเป็นเขียนยอดรวมเวอร์ชันล่าสุด หรือใช้ transaction/deduplication ที่ sink รองรับ

Event ID และ Deduplication

Producer ควรสร้าง Event ID ที่คงเดิมเมื่อ retry ไม่ใช่ UUID ใหม่ทุกครั้ง มิฉะนั้น downstream แยกไม่ออกว่าเป็นเหตุการณ์ใหม่หรือสำเนา State สำหรับ dedup ต้องมีขอบเขตเวลา เพราะการจำ ID ตลอดประวัติทำให้ state โตไม่มีวันจบ ขอบเขตนี้ต้องยาวกว่าระยะที่ duplicate มีโอกาสกลับมา

End-to-End Correctness

ความถูกต้องตลอดสายต้องดู producer, broker, processor และ sink ร่วมกัน ถ้า producer สร้าง event ซ้ำแต่ processor เชื่อว่าทุก ID ใหม่ หรือ sink append อย่างเดียวโดยไม่มี dedup ความรับประกันภายใน Spark ไม่พอ การเขียนเอกสารสถาปัตยกรรมควรระบุ failure scenario แล้วไล่ว่าแต่ละขั้น replay อะไรและผลลัพธ์จะเป็นอย่างไร

11เมื่อยุคเปลี่ยน: จาก DStream สู่ Structured Streaming

Unbounded Input Table และ Result Table

Structured Streaming อธิบาย stream เป็นตาราง input ที่มีแถวใหม่เพิ่มตลอดเวลา Query ทำงานเหมือน DataFrame query แล้วอัปเดต Result Table ตาม event ที่เข้ามา ความคิดนี้ช่วยให้ Filter, Select, GroupBy และ Join มีภาษาเดียวกับ Batch แต่ไม่ได้หมายความว่า operator Batch ทุกตัวใช้กับ Stream ได้โดยไม่มีเงื่อนไข เพราะบาง operation ต้องเห็นข้อมูลทั้งหมดก่อนจึงตอบได้

Output Mode: จะส่งส่วนใดของ Result Table ออกไป

Output Mode ไม่ใช่เพียงรูปแบบการเขียน แต่ผูกกับความหมายว่าแถวใด “เปลี่ยนได้อีก” ตัวอย่าง window aggregate ใน Append Mode ต้องรอ watermark ผ่านจนระบบมองว่า window ไม่ควรได้รับข้อมูลที่ยอมรับเพิ่มแล้ว จึงส่งแถวสุดท้ายออก

Trigger: ประมวลผลบ่อยเพียงใด

val query = counts.writeStream
  .outputMode("update")
  .format("console")
  .option("checkpointLocation", "/checkpoints/event-counts")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()

Trigger 10 วินาทีไม่ได้รับประกันว่าทุกผลลัพธ์ออกภายใน 10 วินาที ถ้า batch ใช้เวลา 25 วินาที trigger ถัดไปไม่สามารถย้อนเวลาแก้กำลังประมวลผลที่ไม่พอได้ Latency ยังรวมเวลารอใน source, scheduling, Shuffle, state store และ sink ด้วย

Continuous Processing ต้องอ่าน Guarantee ให้ครบ

โหมด Continuous Processing ถูกออกแบบเพื่อลด latency อย่างมากสำหรับ query ที่รองรับ แต่มี operator จำกัดและใช้ at-least-once fault-tolerance semantics ต่างจาก default micro-batch ที่รองรับ exactly-once processing semantics ในขอบเขตที่เอกสารกำหนด ดังนั้นประโยค “Continuous เร็วกว่า” ยังไม่พอ ต้องถามต่อว่างานของเรารองรับหรือไม่และ sink รับ duplicate ได้อย่างไร

Stream-Static Join กับ Stream-Stream Join

การ Join stream กับตาราง static เช่น event กับข้อมูลประเภทสินค้า โดยทั่วไปง่ายกว่า เพราะ static side มีขอบเขต แต่ถ้าข้อมูลอ้างอิงเปลี่ยน เราต้องรู้ว่า query จะเห็น version ใหม่เมื่อใด ส่วน Stream-Stream Join ต้องเก็บเหตุการณ์ทั้งสองฝั่งที่ยังรอคู่ จึงต้องมีเงื่อนไขเวลาและ watermark เพื่อกำหนดว่าแถวเก่าเมื่อใดไม่มีคู่ใหม่ที่เป็นไปได้แล้ว

val impressions = impressionStream
  .withWatermark("impression_time", "2 hours")

val clicks = clickStream
  .withWatermark("click_time", "3 hours")

val matched = impressions.as("i").join(
  clicks.as("c"),
  expr("""
    i.ad_id = c.ad_id AND
    c.click_time BETWEEN i.impression_time
                     AND i.impression_time + INTERVAL 1 HOUR
  """)
)

เงื่อนไขเวลาบอกว่า click จะจับคู่กับ impression ได้ภายในหนึ่งชั่วโมง ระบบจึงมีเหตุผลล้าง state เมื่อ watermark ของอีกฝั่งเดินผ่านระยะที่คู่ใหม่เป็นไปไม่ได้ ถ้า Join เพียง ad_id โดยไม่มีขอบเขตเวลา state อาจต้องรอคู่ไปตลอดชีวิตของ query

Streaming Query ไม่ควรอ่านด้วย Code อย่างเดียว

ควรตรวจ query progress ซึ่งมี input rows per second, processed rows per second, batch duration, event-time statistics, watermark และ state operator metrics ถ้า state โตต่อเนื่องทั้งที่ watermark เดิน อาจหมายถึง key cardinality โต, เงื่อนไขล้างไม่ทำงานตามคาด หรือ query ไม่มีขอบเขตที่ระบบใช้ล้างได้

12การวัดและแก้ปัญหา Streaming Pipeline

การปรับ Streaming เริ่มจากแยกอาการ ไม่ใช่เพิ่ม executor ทันที Latency สูงอาจเกิดจาก source lag, query plan, data skew, state store, garbage collection หรือ sink ที่รับไม่ทัน แต่ละสาเหตุต้องแก้คนละจุด

Latency ต้องแยกเป็นช่วง

End-to-end latency อาจแบ่งเป็นเวลา Producer → Broker, รอใน Broker, Broker → Processor, การคำนวณ, State/Checkpoint, Sink write และเวลาที่ผู้ใช้เห็นผล ถ้าวัดเพียง batch duration เราอาจประกาศว่าใช้ 2 วินาที ทั้งที่ event รอใน Kafka มาแล้ว 20 นาที

Throughput กับ Latency ดึงกันคนละทาง

Batch ใหญ่ช่วย amortize scheduling และ I/O overhead ทำให้ throughput ดี แต่ event รอนานขึ้น Batch เล็กตอบเร็วแต่ overhead สูง การออกแบบต้องมี SLO ชัด เช่น 99% ของ event ปรากฏใน dashboard ภายใน 30 วินาทีที่อัตรา 200,000 event/s ไม่ใช่บอกเพียงว่าเป็น Real-time

Load Test ต้องรวม Failure และ Replay

การยิงข้อมูลคงที่ในภาวะปกติไม่พอ ควรทดสอบ burst, broker partition หยุด, executor หลุด, sink ช้า, restart จาก checkpoint, schema เปลี่ยน และ replay ข้อมูลเก่า ระบบอาจรับ live traffic ได้ดีแต่พังกู้คืน เพราะต้องประมวลผล backlog พร้อม traffic ใหม่

Observability ของข้อมูลกับระบบเป็นคนละชุด

Pipeline อาจเขียวทุก dashboard ทางระบบ แต่ข้อมูล region หนึ่งหายไปทั้งวัน การ monitor เพียง infrastructure จึงตอบไม่ได้ว่าผลลัพธ์ถูกต้อง

13กรณีศึกษา: ตรวจจับธุรกรรมผิดปกติแบบต่อเนื่อง

สมมุติธนาคารต้องแจ้งเตือนเมื่อบัญชีมีธุรกรรมจำนวนมากผิดปกติภายในช่วงสั้น พร้อมตรวจ pattern การเข้าสู่ระบบจากประเทศใหม่ โจทย์นี้รวม Window, State, CEP, Deduplication และ ML ไว้ด้วยกัน

ขั้นที่ 1: กำหนด Event Contract

แต่ละ event มี event_id, account_id, event_time, ingestion_time, amount, country, device และ schema version Event ID คงเดิมเมื่อ producer retry ส่วน timestamp ใช้ UTC และมี rule แยกค่าที่อยู่ไกลเกินจริง

ขั้นที่ 2: Deduplicate ก่อนสร้าง State ธุรกิจ

ระบบตัด duplicate ตาม Event ID ภายในช่วงเวลาที่ครอบคลุม retry ที่คาดไว้ ถ้าไม่ทำ จำนวนธุรกรรมและ feature จะสูงเกินจริง แต่ state ของ ID ก็ต้องมี watermark/TTL มิฉะนั้นโตตามจำนวนธุรกรรมตลอดไป

ขั้นที่ 3: สร้าง Feature หลาย Window

• จำนวนธุรกรรมและยอดรวมใน 1 นาที เพื่อจับ burst
• จำนวนประเทศและอุปกรณ์ใน 1 ชั่วโมง เพื่อจับพฤติกรรมเปลี่ยน
• ค่าเฉลี่ยและส่วนเบี่ยงเบนจากประวัติระยะยาวเพื่อเทียบ baseline

Window สั้นต้องการความสด ส่วน baseline ระยะยาวอาจใช้ summary หรือ precomputed profile ไม่จำเป็นต้องเก็บ raw event ทุกตัวไว้ใน state เดียว

ขั้นที่ 4: Rule กับ Model ทำงานคนละหน้าที่

กฎ เช่น “ผิด PIN 5 ครั้งแล้วโอนเงินภายใน 3 นาที” เหมาะกับ CEP เพราะลำดับมีความหมาย ส่วนโมเดลให้ risk score จาก feature หลายมิติ Rule ที่มีความเสี่ยงสูงอาจ block ทันที ขณะที่ score ปานกลางส่งให้ตรวจสอบ การรวมสองแบบช่วยให้มีทั้ง pattern ที่อธิบายได้และความยืดหยุ่นทางสถิติ

ขั้นที่ 5: Sink ต้องไม่สร้าง Alert ซ้ำ

สร้าง Alert ID จาก account, rule/model version และช่วงเวลา แล้ว upsert ไปยังระบบ case management ถ้า processor retry alert เดิมจะไม่เปิดคดีซ้ำ การส่ง SMS หรือระงับบัญชีควรใช้ idempotency key และมี audit log แยก

ขั้นที่ 6: ผลลัพธ์มาช้าก็ยังมีประโยชน์

ธุรกรรมที่มาช้าเกิน watermark อาจไม่เปลี่ยน alert แบบทันที แต่ส่งเข้า late-event table เพื่อ reconciliation และปรับปรุง model data ภายหลัง เราแยกเส้นทางตอบสนองเร็วออกจากเส้นทางบัญชีที่ต้องครบ แทนการบังคับให้ระบบเดียวตอบสอง SLA ที่ขัดกัน

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

Streaming เปลี่ยนโจทย์จากการคำนวณข้อมูลที่รู้ขอบเขต ไปเป็นการตัดสินใจจากข้อมูลที่ยังมาไม่ครบ เราจึงต้องแยก Event Time ออกจาก Processing Time เลือก Window ตามความหมายของคำถาม ใช้ Watermark เพื่อสร้างขอบเขตการรอ และกำหนด lifecycle ของ State อย่างชัดเจน

Smart Window Counting แสดงให้เห็นอีกมุมหนึ่งว่าเราไม่จำเป็นต้องเก็บอดีตทุกเหตุการณ์เพื่อให้ตอบคำถามได้ ถ้าโจทย์ยอมรับ approximation เราสามารถรักษา summary ที่ใช้พื้นที่เล็กมาก เช่น DGIM สำหรับนับเลข 1 ใน sliding window หลักสำคัญไม่ใช่จำชื่อ Algorithm แต่คือรู้ว่า summary เก็บอะไร สูญเสียอะไร รวมข้ามเครื่องได้หรือไม่ และ error ที่ยอมรับได้เท่าไร

ขั้นตอนถัดไป คือข้อมูลแบบกราฟและ PageRank เราจะเปลี่ยนจากเหตุการณ์ที่เรียงตามเวลาไปสู่ข้อมูลที่ความสัมพันธ์ระหว่างโหนดเป็นหัวใจ แม้โจทย์ต่างกัน แต่แนวคิดเดิมยังตามไปด้วย ได้แก่การแบ่งข้อมูล การลดการเคลื่อนย้าย State การคำนวณซ้ำแบบ incremental และการแลกความแม่นกับทรัพยากรในระบบขนาดใหญ่