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

John Gall, Systemantics (1975)

ใน บทที่ 11 เราได้พูดคุยเกี่ยวกับ batch processing ซึ่งเป็นเทคนิคที่อ่านชุดไฟล์เป็น input และสร้างชุดไฟล์ output ใหม่ Output ที่ได้นั้นเป็นรูปแบบหนึ่งของ derived data หรือก็คือชุดข้อมูลที่สามารถสร้างขึ้นใหม่ได้โดยการรัน batch process ซ้ำอีกครั้งหากจำเป็น เราได้เห็นแล้วว่าแนวคิดที่เรียบง่ายแต่ทรงพลังนี้สามารถใช้สร้าง search indexes, recommendation systems, analytics และอื่นๆ ได้อย่างไร

อย่างไรก็ตาม ยังมีข้อสมมติฐานใหญ่ข้อหนึ่งที่ใช้มาตลอดทั้ง บทที่ 11 : input มีขอบเขตจำกัด คือมีขนาดแน่นอนและรู้ล่วงหน้า batch process จึงรู้ว่าอ่าน input เสร็จเมื่อไหร่ ตัวอย่างเช่น การดำเนินการ sort ที่เป็นหัวใจของ MapReduce ต้องอ่าน input ทั้งหมดก่อนจึงจะเริ่มสร้าง output ได้ เพราะ record สุดท้ายของ input อาจมี key ต่ำที่สุด ซึ่งต้องไปอยู่เป็น record แรกของ output จึงเริ่มสร้าง output ก่อนไม่ได้

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

ปัญหาของ daily batch processes คือกว่าการเปลี่ยนแปลงใน input จะไปปรากฏใน output ต้องรอนานกว่าหนึ่งวัน ซึ่งช้าเกินไปสำหรับผู้ใช้ที่ใจร้อนจำนวนมาก เพื่อลดความล่าช้า เราสามารถรันการประมวลผลให้บ่อยขึ้น — เช่น การประมวลผลข้อมูลหนึ่งวินาทีเมื่อสิ้นสุดแต่ละวินาที — หรือแม้กระทั่งทำอย่างต่อเนื่อง โดยไม่แบ่งเวลาเป็นช่วงตายตัว และประมวลผลทุก event ทันทีที่เกิดขึ้น นี่คือแนวคิดเบื้องหลัง stream processing (การประมวลผลสตรีม)

โดยทั่วไปแล้ว stream หมายถึงข้อมูลที่ถูกทำให้พร้อมใช้งานเพิ่มขึ้นเรื่อยๆ เมื่อเวลาผ่านไป แนวคิดนี้ปรากฏในหลายที่: ใน stdin และ stdout ของ Unix, ภาษาโปรแกรม (lazy lists) [ 2 ], filesystem APIs (เช่น FileInputStream ของ Java), การเชื่อมต่อ TCP, เสียงและวิดีโอที่ส่งผ่านอินเทอร์เน็ต และอื่นๆ

ในบทนี้เราจะดู event streams (สตรีมเหตุการณ์) ในฐานะกลไกจัดการข้อมูล: เป็นข้อมูลที่ไม่มีขอบเขตและประมวลผลทีละส่วน ซึ่งเทียบได้กับข้อมูลแบบ batch ในบทที่แล้ว เริ่มจากดูว่า stream ถูกแทนด้วยรูปแบบใด เก็บอย่างไร และส่งผ่านเครือข่ายอย่างไร จากนั้นจะสำรวจ ความสัมพันธ์ ระหว่าง stream และฐานข้อมูล สุดท้ายในหัวข้อ "Processing Streams" เราจะสำรวจแนวทางและเครื่องมือสำหรับการประมวลผล stream เหล่านั้น อย่างต่อเนื่อง และวิธีที่สามารถนำมาใช้สร้างแอปพลิเคชันได้

Transmitting Event Streams (การส่งผ่านสตรีมเหตุการณ์)

ใน โลกของ batch processing input และ output ของ job คือไฟล์ (อาจจะอยู่บน distributed filesystem) แล้วสิ่งที่เทียบเท่าในโลก streaming ล่ะ?

เมื่อ input เป็นไฟล์ (ลำดับของ bytes) ขั้นตอนแรกโดยปกติคือ parse เป็นลำดับของ record ในบริบทของ stream processing record มักถูกเรียกว่า event (เหตุการณ์) แต่โดยพื้นฐานแล้วมันคือสิ่งเดียวกัน: object ขนาดเล็กที่มีข้อมูลครบในตัวเองและเปลี่ยนแปลงไม่ได้ ซึ่งบันทึกรายละเอียดว่ามีอะไรเกิดขึ้น ณ จุดเวลาใดเวลาหนึ่ง Event มักจะประกอบด้วย timestamp ที่ระบุว่าเกิดขึ้นเมื่อไหร่ตาม time-of-day clock (ดู "Monotonic Versus Time-of-Day Clocks" )

ตัวอย่างเช่น สิ่งที่เกิดขึ้นอาจเป็นการกระทำที่ผู้ใช้ทำ เช่น การดูหน้าเว็บหรือการซื้อสินค้า มันอาจมาจากเครื่องจักร เช่น การวัดค่าเป็นระยะจากเซนเซอร์อุณหภูมิหรือค่า CPU utilization ในตัวอย่างในหัวข้อ "Batch Processing with Unix Tools" แต่ละบรรทัดของ web server log ถือเป็นหนึ่ง event

Event อาจถูกเข้ารหัสเป็น text string, JSON หรือรูปแบบ binary ดังที่กล่าวถึงใน บทที่ 5 การเข้ารหัสนี้ทำให้คุณสามารถจัดเก็บ event ได้ — เช่น โดยการต่อท้ายลงในไฟล์, แทรกเข้าไปในตาราง relational, หรือเขียนลงใน document database การเข้ารหัสยังช่วยให้คุณส่ง event ผ่านเครือข่ายไปยังโหนดอื่นเพื่อประมวลผลได้อีกด้วย

ในแบบ batch processing ไฟล์ถูกเขียนครั้งเดียวแล้วอาจถูกอ่านโดยหลาย jobs ในทำนองเดียวกัน ในคำศัพท์ของ streaming event ถูกสร้างขึ้นครั้งเดียวโดย producer (หรือที่เรียกว่า publisher หรือ sender ) แล้วอาจถูกประมวลผลโดย consumers หลายตัว ( subscribers หรือ recipients ) [ 3 ] ใน filesystem ชื่อไฟล์ระบุชุดของ records ที่เกี่ยวข้องกัน ใน streaming system events ที่เกี่ยวข้องกันมักถูกจัดกลุ่มเข้าด้วยกันเป็น topic หรือ stream

ในหลักการ เราใช้ไฟล์หรือฐานข้อมูลเชื่อม producer กับ consumer ได้เลย Producer เขียนทุก event ที่มันสร้างลงใน datastore และแต่ละ consumer จะคอย poll datastore เป็นระยะ เพื่อดูว่ามี event ใหม่เกิดขึ้นหลังจากรอบที่แล้วหรือไม่ โดยพื้นฐานแล้วนี่คือสิ่งที่ batch process ทำเมื่อประมวลผลข้อมูลหนึ่งวันเมื่อสิ้นสุดแต่ละวัน

อย่างไรก็ตาม เมื่อขยับไปสู่การประมวลผลต่อเนื่องที่ latency ต่ำ การ poll จะเปลืองต้นทุน ถ้า datastore ไม่ได้ออกแบบมาสำหรับการใช้งานลักษณะนี้ ยิ่ง poll บ่อย สัดส่วนของ request ที่ได้ event ใหม่กลับมาก็ยิ่งน้อยลง และ overhead ก็ยิ่งสูงขึ้น จึงดีกว่าถ้าระบบแจ้ง consumer เมื่อมี event ใหม่เกิดขึ้น

ฐานข้อมูลส่วนใหญ่ไม่ค่อยรองรับกลไกการแจ้งเตือนแบบนี้ relational database มักมี triggers ซึ่งทำงานตอบสนองต่อการเปลี่ยนแปลงได้ (เช่น เมื่อแทรกแถวใหม่ลงในตาราง) แต่ trigger ทำอะไรได้จำกัดมาก และเหมือนเป็นฟีเจอร์ที่เพิ่มเข้ามาทีหลังในการออกแบบฐานข้อมูล [ 4 ] จึงมีเครื่องมือเฉพาะทางที่พัฒนาขึ้นมาเพื่อส่งการแจ้งเตือน event โดยเฉพาะ

Messaging Systems (ระบบรับส่งข้อความ)

แนวทางทั่วไปในการแจ้งเตือน consumers เกี่ยวกับ events ใหม่คือการใช้ messaging system (ระบบรับส่งข้อความ): producer ส่ง message ที่บรรจุ event ไว้ แล้วระบบจะส่งต่อให้ consumer เราได้กล่าวถึงระบบเหล่านี้ก่อนหน้านี้ในหัวข้อ "Event-Driven Architectures" และตอนนี้เราจะลงรายละเอียดเพิ่มเติม

ช่องทางการสื่อสารโดยตรง เช่น Unix pipe หรือ TCP connection ระหว่าง producer และ consumer จะเป็นวิธีง่ายๆ ในการ implement messaging system อย่างไรก็ตาม messaging systems ส่วนใหญ่ขยายจากโมเดลพื้นฐานนี้ โดยเฉพาะอย่างยิ่ง Unix pipes และ TCP เชื่อมต่อ sender หนึ่งรายกับ recipient หนึ่งรายอย่างเจาะจง ในขณะที่ messaging system ให้ producer node หลายตัวส่ง message ไปยัง topic เดียวกันได้ และให้ consumer node หลายตัวรับ message ใน topic นั้นได้

ภายใต้โมเดล publish/subscribe นี้ แต่ละระบบมีแนวทางต่างกันมาก และไม่มีคำตอบเดียวที่เหมาะกับทุกงาน เวลาเปรียบเทียบระบบเหล่านี้ คำถามสองข้อต่อไปนี้ช่วยได้มาก:

จะเกิดอะไรขึ้นถ้า producer ส่ง message เร็วกว่าที่ consumer จะประมวลผลทัน?

โดยกว้างๆ แล้ว ระบบมีสามทางเลือก: drop message ทิ้ง, buffer message ไว้ในคิว, หรือใช้ backpressure (หรือที่เรียกว่า flow control ซึ่งจะบล็อกไม่ให้ producer ส่ง message เพิ่ม) ตัวอย่างเช่น Unix pipes และ TCP ใช้ backpressure; buffer ของมันมีขนาดคงที่และเล็ก ถ้า buffer เต็ม ฝั่งส่งจะถูกบล็อกจนกว่าฝั่งรับจะดึงข้อมูลออกไป (ดู "Network congestion and queueing" )

ถ้า message ถูกเก็บไว้ในคิว ต้องเข้าใจด้วยว่าจะเกิดอะไรขึ้นเมื่อคิวนั้นโตขึ้นเรื่อยๆ ระบบจะ crash ไหมถ้าคิวใหญ่เกินกว่าจะอยู่ใน memory หรือจะเขียน message ลง disk แทน? ถ้าเป็นแบบหลัง การเข้าถึง disk กระทบประสิทธิภาพของ messaging system อย่างไร [ 5 ] แล้วจะเกิดอะไรขึ้นเมื่อ disk เต็ม [ 6 ]?

จะเกิดอะไรขึ้นถ้า node crash หรือออฟไลน์ชั่วคราว — message จะสูญหายไหม?

เช่นเดียวกับฐานข้อมูล durability (ความทนทาน) อาจต้องอาศัยการเขียนลง disk และ/หรือ replication ผสมกัน (ดู sidebar "Replication and Durability" ) ซึ่งมีต้นทุน ถ้ายอมให้ message สูญหายได้บ้าง ก็อาจได้ throughput สูงขึ้นและ latency ต่ำลงบนฮาร์ดแวร์เดียวกัน

การปล่อยให้ message สูญหายได้หรือไม่ ขึ้นอยู่กับแอปพลิเคชันเป็นอย่างมาก ตัวอย่างเช่น กับ sensor readings และ metrics ที่ถูกส่งเป็นระยะ ข้อมูลที่ขาดหายไปเป็นครั้งคราวอาจไม่สำคัญ เพราะค่าที่อัปเดตจะถูกส่งมาในไม่ช้า แต่ต้องระวังว่าถ้า message จำนวนมากถูก drop ทิ้ง เราอาจไม่ทันสังเกตว่า metrics ผิดเพี้ยน [ 7 ] ถ้าคุณกำลังนับ events การส่งมอบอย่างเชื่อถือได้สำคัญกว่า เพราะทุก message ที่หายไปทำให้ตัวนับผิด

ข้อดีอย่างหนึ่งของ batch processing system ที่เราดูใน บทที่ 11 คือให้การรับประกันด้านความน่าเชื่อถือที่แข็งแกร่ง task ที่ล้มเหลวจะถูก retry อัตโนมัติ และ output บางส่วนจาก task ที่ล้มเหลวจะถูกทิ้งไปเอง ทำให้ output ออกมาเหมือนไม่เคยมีข้อผิดพลาดเกิดขึ้น ซึ่งช่วยให้โมเดลการเขียนโปรแกรมง่ายขึ้น ต่อไปในบทนี้เราจะดูว่าจะให้การรับประกันแบบเดียวกันในบริบทของ streaming ได้อย่างไร

Direct messaging from producers to consumers (การส่งข้อความโดยตรงจาก producer ไปยัง consumer)

มี messaging systems จำนวนหนึ่งที่ใช้การสื่อสารผ่านเครือข่ายโดยตรงระหว่าง producers และ consumers โดยไม่ต้องผ่าน intermediary nodes:

• UDP multicast ถูกใช้อย่างแพร่หลายในอุตสาหกรรมการเงินสำหรับ streams เช่น stock market feeds ที่ต้องการ latency ต่ำ [ 8 ] แม้ว่า UDP เองจะไม่น่าเชื่อถือ แต่โปรโตคอลระดับแอปพลิเคชันสามารถกู้คืนแพ็กเก็ตที่สูญหายได้ (producer ต้องจำ packet ที่ส่งไปแล้ว เพื่อส่งซ้ำได้เมื่อมีการร้องขอ)

• ไลบรารี messaging ที่ไม่มี broker อย่าง ZeroMQ และ nanomsg ใช้แนวทางที่คล้ายกัน โดย implement publish/subscribe messaging ผ่าน TCP หรือ IP multicast

• ตัวเก็บ metrics บางตัว เช่น StatsD [ 9 ] ใช้ UDP messaging ที่ไม่น่าเชื่อถือเพื่อเก็บ metrics จากทุกเครื่องบนเครือข่ายแล้วนำไป monitor (ในโปรโตคอล StatsD ตัวนับ metrics จะถูกต้องก็ต่อเมื่อได้รับ message ครบทุกตัว; การใช้ UDP จึงทำให้ metrics แม่นยำแค่โดยประมาณ [ 10 ] ดูเพิ่มเติมที่ "TCP Versus UDP" )

• ถ้า consumer เปิด service บนเครือข่าย producers สามารถส่ง HTTP หรือ RPC request โดยตรง (ดู "Dataflow Through Services: REST and RPC" ) เพื่อ push message ไปยัง consumer นี่คือแนวคิดเบื้องหลัง webhooks [ 11 ] ซึ่งเป็น pattern ที่ service หนึ่งลงทะเบียน callback URL ไว้กับอีก service หนึ่ง แล้ว service หลังจะส่ง request ไปยัง URL นั้นเมื่อมี event เกิดขึ้น

ระบบ direct messaging เหล่านี้ทำงานได้ดีในสถานการณ์ที่ออกแบบมาให้ใช้ แต่โดยทั่วไปโค้ดแอปพลิเคชันต้องคำนึงถึงโอกาสที่ message จะสูญหายด้วย และ fault ที่ทนได้ก็ค่อนข้างจำกัด แม้โปรโตคอลจะตรวจจับและส่ง packet ที่หายไปในเครือข่ายซ้ำได้ แต่โดยทั่วไปมันสมมติว่า producer และ consumer ออนไลน์อยู่ตลอดเวลา

ถ้า consumer ออฟไลน์ ก็อาจพลาด message ที่ถูกส่งมาในช่วงที่ติดต่อไม่ได้ บางโปรโตคอลให้ producer retry message ที่ส่งไม่สำเร็จได้ แต่ถ้า producer crash แล้ว buffer ของ message ที่ต้องส่งซ้ำหายไป วิธีนี้ก็อาจใช้ไม่ได้

Message brokers (ตัวกลางรับส่งข้อความ)

อีกทางเลือกที่ใช้กันแพร่หลายคือส่ง message ผ่าน message broker (หรือที่เรียกว่า message queue ) ซึ่งโดยพื้นฐานคือฐานข้อมูลชนิดหนึ่งที่ปรับมาเพื่อจัดการ stream ของ message [ 12 ] broker ทำงานเป็น server โดยมี producer และ consumer เชื่อมต่อเข้ามาในฐานะ client Producer เขียน message ไปที่ broker แล้ว broker ส่ง message เหล่านั้นต่อให้ consumer

เมื่อรวมศูนย์ข้อมูลไว้ที่ broker ระบบเหล่านี้จึงรับมือกับ client ที่เข้าๆ ออกๆ (เชื่อมต่อ, ตัดการเชื่อมต่อ, และ crash) ได้ง่ายขึ้น และเรื่อง durability ก็กลายเป็นหน้าที่ของ broker แทน broker บางตัวเก็บ message ไว้ใน memory เท่านั้น ส่วนบางตัว (ขึ้นอยู่กับ config) เขียนลง disk เพื่อไม่ให้ message หายเมื่อ broker crash เมื่อ consumer ช้า broker เหล่านี้มักยอมให้ queue โตได้ไม่จำกัด (แทนที่จะ drop message หรือใช้ backpressure) แม้ตัวเลือกนี้ก็อาจขึ้นอยู่กับ config เช่นกัน

ผลที่ตามมาของ queueing คือ consumer มักทำงานแบบ asynchronous (ไม่ประสานเวลา) เมื่อ producer ส่ง message โดยปกติจะรอแค่ให้ broker ยืนยันว่า buffer message ไว้แล้ว ไม่ได้รอให้ consumer ประมวลผล message message จะถูกส่งให้ consumer ในอนาคต ณ เวลาที่ไม่แน่นอน — โดยปกติภายในเสี้ยววินาที แต่บางครั้งอาจช้ากว่านั้นมากถ้ามี queue backlog

Message brokers compared to databases (การเปรียบเทียบ message brokers กับฐานข้อมูล)

broker บางตัวถึงขั้นเข้าร่วม two-phase commit protocol ได้โดยใช้ XA หรือ JTA (ดู "Distributed Transactions Across Different Systems" ) ซึ่งทำให้มันคล้ายฐานข้อมูลมากขึ้น แต่ในทางปฏิบัติ message broker กับฐานข้อมูลก็ยังต่างกันในประเด็นสำคัญหลายอย่าง:

• ฐานข้อมูลมักเก็บข้อมูลไว้จนกว่าจะถูกลบอย่างชัดเจน ส่วน message broker บางตัว จะลบ message ทิ้งอัตโนมัติเมื่อส่งให้ consumer สำเร็จแล้ว broker แบบนี้จึงไม่เหมาะกับการเก็บข้อมูลระยะยาว

• เพราะลบ message เร็ว broker ส่วนใหญ่จึงสมมติว่า working set ของมันค่อนข้างเล็ก คือ queue สั้น ถ้า broker ต้อง buffer message จำนวนมากเพราะ consumer ช้า (และอาจ spill message ลง disk เมื่อ memory ไม่พอ) แต่ละ message จะใช้เวลาประมวลผลนานขึ้น และ throughput โดยรวมอาจลดลง [ 5 ]

• ฐานข้อมูลมักรองรับ secondary index และวิธีค้นหาข้อมูลต่างๆ ด้วย query language ส่วน message broker มักให้ subscribe เฉพาะ topic บางส่วน ที่ตรงกับ pattern ที่กำหนด ทั้งสองแบบโดยพื้นฐานเป็นวิธีให้ client เลือกส่วนของข้อมูลที่ตัวเองสนใจ แต่ฐานข้อมูลมีความสามารถในการค้นหาที่ก้าวหน้ากว่ามาก

• ผลลัพธ์ของ query ฐานข้อมูลมักอิงกับ point-in-time snapshot ของข้อมูล ถ้ามี client อื่นเขียนข้อมูลที่ทำให้ผลของ query เปลี่ยนไป client แรกก็จะไม่รู้ว่าผลลัพธ์ที่ได้ไปก่อนหน้านี้ล้าสมัยแล้ว (เว้นแต่จะ query ซ้ำหรือ poll เพื่อดูการเปลี่ยนแปลง) ในทางกลับกัน message broker ไม่รองรับ query แบบ arbitrary และไม่ให้แก้ไข message หลังจากส่งไปแล้ว แต่จะแจ้ง client เมื่อข้อมูลเปลี่ยนแปลง (เช่น เมื่อมี message ใหม่เข้ามา)

นี่ คือมุมมองดั้งเดิมของ message broker ซึ่งถูกกำหนดไว้ในมาตรฐานอย่าง JMS [ 13 ] และ AMQP [ 14 ] และ implement อยู่ในซอฟต์แวร์อย่าง RabbitMQ, ActiveMQ, HornetQ, Qpid, TIBCO Enterprise Message Service, IBM MQ, Azure Service Bus, และ Google Cloud Pub/Sub [ 15 ] แม้จะใช้ฐานข้อมูลเป็น queue ได้ แต่การ tune ให้ได้ประสิทธิภาพดีไม่ใช่เรื่องตรงไปตรงมา [ 16 ]

Multiple consumers (การมี consumer หลายตัว)

เมื่อมี consumer หลายตัวอ่าน message จาก topic เดียวกัน จะมี pattern หลักสองแบบในการส่ง message ดังที่แสดงใน Figure 12-1 :

Load balancing (การกระจายโหลด)

แต่ละ message ถูกส่งไปยัง หนึ่ง ใน consumer ดังนั้น consumer จึงช่วยกันแบ่งงานประมวลผล message ใน topic ได้ Broker อาจเลือกส่ง message ให้ consumer ตัวไหนก็ได้ Pattern นี้มีประโยชน์เมื่อแต่ละ message ใช้ต้นทุนประมวลผลสูง คุณจึงอยากเพิ่ม consumer มาประมวลผลแบบขนาน (ใน AMQP คุณ implement load balancing ได้โดยให้ client หลายตัว consume จาก queue เดียวกัน ส่วนใน JMS เรียกว่า shared subscription )

Fan-out (การกระจายไปยังทุก consumer)

แต่ละ message ถูกส่งไปยัง ทั้งหมด ของ consumer Fan-out ทำให้ consumer อิสระหลายตัว "ฟัง" broadcast ของ message ชุดเดียวกันได้โดยไม่กระทบกัน — ซึ่งในโลก streaming ก็เทียบได้กับการมี batch job หลายตัวอ่านไฟล์ input เดียวกัน (JMS ให้ฟีเจอร์นี้ผ่าน topic subscription และ AMQP ผ่าน exchange binding)

Figure 12-1. (a) Load balancing แบ่งงาน consume topic ให้ consumer หลายตัว; (b) Fan-out ส่งแต่ละ message ไปให้ consumer หลายตัว

สอง pattern นี้ใช้ร่วมกันได้ เช่น ฟีเจอร์ consumer groups ของ Kafka เมื่อ consumer group หนึ่ง subscribe topic ใด แต่ละ message ใน topic จะถูกส่งให้ consumer หนึ่งตัวใน group (กระจายโหลดไปทั่วทั้ง group) ถ้ามี consumer group แยกกันสองกลุ่ม subscribe topic เดียวกัน แต่ละ message จะถูกส่งให้ consumer หนึ่งตัวในแต่ละ group (เกิด fan-out ข้าม group)

Acknowledgments and redelivery (การยืนยันและการส่งซ้ำ)

Consumer อาจ crash ได้ทุกเมื่อ จึงอาจเกิดกรณีที่ broker ส่ง message ให้ consumer แต่ consumer ไม่เคยประมวลผลเลย หรือประมวลผลไปเพียงบางส่วนก่อน crash เพื่อไม่ให้ message สูญหาย message broker จึงใช้ acknowledgments (การยืนยัน): client ต้องบอก broker อย่างชัดเจนเมื่อประมวลผล message เสร็จแล้ว broker จะได้ลบ message ออกจาก queue

ถ้า connection กับ client ถูกปิดหรือ timeout โดยที่ broker ยังไม่ได้รับ ack broker จะถือว่า message นั้นยังไม่ได้ถูกประมวลผล จึงส่งให้ consumer อื่นอีกครั้ง (ข้อควรระวังคือ message นั้นอาจ ถูก ประมวลผลเสร็จไปแล้ว แต่ ack สูญหายระหว่างทางในเครือข่าย การจัดการกรณีนี้ต้องใช้ atomic commit protocol ดังที่กล่าวไว้ในหัวข้อ "Exactly-once message processing" เว้นแต่การดำเนินการนั้นจะ idempotent หรือไม่ต้องการ exactly-once semantics)

เมื่อใช้ร่วมกับ load balancing พฤติกรรม redelivery นี้ส่งผลที่น่าสนใจต่อลำดับของ message ใน Figure 12-2 โดยปกติ consumer จะประมวลผล message ตามลำดับที่ producer ส่งมา แต่ consumer 2 crash ขณะประมวลผล message m3 ในขณะที่ consumer 1 กำลังประมวลผล message m4 message m3 ที่ยังไม่ได้ ack จะถูกส่งซ้ำให้ consumer 1 ทำให้ consumer 1 ประมวลผล message ตามลำดับ m4 , m3 , m5 ทำให้ m3 และ m4 มาถึงไม่ตรงลำดับที่ producer 1 ส่ง

Figure 12-2. Consumer 2 crash ขณะประมวลผล m3 จึงมีการส่ง m3 ซ้ำให้ consumer 1 ในภายหลัง

แม้ message broker จะพยายามรักษาลำดับของ message (ตามที่มาตรฐาน JMS และ AMQP กำหนด) แต่การใช้ load balancing ร่วมกับ redelivery ย่อมทำให้ลำดับของ message ถูกสลับ เพื่อเลี่ยงปัญหานี้ คุณใช้ queue แยกสำหรับแต่ละ consumer ได้ (คือไม่ใช้ load balancing) การที่ลำดับสลับกันไม่ใช่ปัญหา ถ้า message เป็นอิสระต่อกันโดยสมบูรณ์ แต่อาจเป็นปัญหาได้ถ้า message มีความสัมพันธ์เชิงเหตุและผลต่อกัน ดังที่เราจะเห็นต่อไปในบทนี้

redelivery ยังอาจทำให้สิ้นเปลืองทรัพยากร, เกิด resource starvation, หรือทำให้ stream ติดค้างถาวรได้ กรณีที่พบบ่อยคือ producer serialize message ผิดพลาด — เช่น ลืมใส่ key ที่จำเป็นใน object ที่เข้ารหัสเป็น JSON ถ้า message ที่ขาด key ทำให้ consumer crash แล้ว restart consumer จะไม่ได้ ack message นั้น broker จึงส่งซ้ำ ซึ่งทำให้ consumer ตัวอื่นล้มเหลวอีก วงจรนี้จะวนซ้ำไปเรื่อยๆ ไม่มีวันจบ ถ้า broker รับประกันลำดับอย่างเข้มงวด ระบบจะไม่คืบหน้าต่อไปได้อีก Broker ที่ยอมให้ลำดับ message สลับได้ยังคืบหน้าต่อไปได้ แต่จะเสียทรัพยากรไปกับ message ที่ไม่มีทางถูก ack

ปัญหานี้จัดการได้ด้วย dead letter queue (DLQ) แทนที่จะเก็บ message ไว้ใน queue เดิมแล้ว retry ไปเรื่อยๆ ก็ย้าย message ไปไว้ใน queue อื่นเพื่อปลดบล็อก consumer [ 17 , 18 ] โดยปกติจะตั้ง monitoring บน DLQ — message ใดๆ ใน queue ถือเป็นข้อผิดพลาด เมื่อพบ message ใหม่ operator อาจเลือก drop ทิ้งถาวร, แก้ไขแล้วสร้าง message ใหม่เอง หรือแก้โค้ด consumer ให้จัดการ message นั้นได้อย่างเหมาะสม ระบบ queuing ส่วนใหญ่มี DLQ และระบบ messaging แบบ log-based เช่น Apache Pulsar กับระบบ stream processing เช่น Kafka Streams ก็รองรับเช่นกัน [ 19 ]

Log-Based Message Brokers (message broker แบบ log-based)

การส่ง packet ผ่านเครือข่ายหรือการส่ง request ไปยัง network service มักเป็นการกระทำชั่วคราว ที่ไม่ทิ้งร่องรอยถาวรไว้ แม้จะบันทึกการกระทำเหล่านี้ไว้ถาวรได้ (โดยใช้ packet capture และ logging) แต่ปกติเราไม่ได้คิดกันแบบนั้น message broker แบบ AMQP/JMS ก็สืบทอดแนวคิดการส่ง message แบบชั่วคราวนี้มา แม้อาจเขียน message ลง disk แต่ก็ลบทิ้งอย่างรวดเร็วหลังส่งให้ consumer แล้ว

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

ความต่างทางแนวคิดนี้ส่งผลมากต่อวิธีสร้าง derived data คุณสมบัติสำคัญของ batch process ตามที่กล่าวไว้ใน บทที่ 11 คือรันซ้ำได้ ลองปรับขั้นตอนการประมวลผลได้ โดยไม่เสี่ยงทำลาย input (เพราะ input เป็น read-only) การรับส่ง message แบบ AMQP/JMS ไม่เป็นเช่นนั้น: การรับ message เป็นการกระทำที่ทำลายข้อมูล ถ้า acknowledgment ทำให้ message ถูกลบออกจาก broker เราจึงรัน consumer ตัวเดิมซ้ำแล้วคาดหวังผลลัพธ์เหมือนเดิมไม่ได้

ถ้าเพิ่ม consumer ใหม่เข้า messaging system โดยปกติมันจะเริ่มรับเฉพาะ message ที่ส่งหลังจากลงทะเบียนแล้วเท่านั้น message ก่อนหน้านั้นหายไปแล้วและกู้คืนไม่ได้ ต่างจากไฟล์และฐานข้อมูลที่เพิ่ม client ใหม่เมื่อไหร่ก็ได้ และ client นั้นย้อนอ่านข้อมูลที่เขียนไว้ในอดีตได้ไกลเท่าที่ต้องการ (ตราบใดที่แอปพลิเคชันยังไม่ได้เขียนทับหรือลบข้อมูลนั้น)

ทำไมเราไม่ทำระบบแบบผสม ที่รวมข้อดีของ durable storage แบบฐานข้อมูล เข้ากับการแจ้งเตือนแบบ latency ต่ำของ messaging? นี่คือแนวคิดเบื้องหลัง log-based message brokers ซึ่งเป็นที่นิยมมากในช่วงไม่กี่ปีที่ผ่านมา

Using logs for message storage (การใช้ log เก็บ message)

log คือลำดับของ record บน disk ที่เขียนต่อท้ายได้อย่างเดียว (append-only) เราเคยพูดถึง log มาแล้วในบริบทของ log-structured storage engine และ write-ahead log ใน บทที่ 4 , ในบริบทของ replication ใน บทที่ 6 , และในรูปแบบของ consensus ใน บทที่ 10

เราใช้โครงสร้างเดียวกันนี้ implement message broker ได้ producer ส่ง message โดยเขียนต่อท้าย log และ consumer รับ message โดยอ่าน log ตามลำดับ ถ้า consumer อ่านถึงท้าย log แล้ว ก็จะรอการแจ้งเตือนว่ามี message ใหม่ถูกเขียนต่อท้าย คำสั่ง Unix tail เมื่อใส่ตัวเลือก -f ซึ่งเฝ้าดูไฟล์เพื่อรอข้อมูลที่ถูกเขียนต่อท้าย โดยพื้นฐานแล้วก็ทำงานแบบนี้

เพื่อรองรับ throughput ที่สูงกว่าที่ disk ตัวเดียวทำได้ log สามารถถูก sharded (แบ่งเป็นส่วน) ได้ (ในความหมายของ บทที่ 7 ) แต่ละ shard จะโฮสต์บนเครื่องคนละเครื่องก็ได้ ทำให้แต่ละ shard เป็น log แยกกัน ที่อ่านและเขียนได้อิสระจาก shard อื่น ส่วน topic ก็คือกลุ่มของ shard ที่เก็บ message ประเภทเดียวกันทั้งหมด แนวทางนี้แสดงใน Figure 12-3

Figure 12-3. Producer ส่ง message โดยเขียนต่อท้ายไฟล์ของ topic partition และ consumer อ่านไฟล์เหล่านี้ตามลำดับ

ภายในแต่ละ shard ซึ่ง Kafka เรียกว่า partition , broker จะกำหนดหมายเลขลำดับที่เพิ่มขึ้นเรื่อยๆ (monotonically increasing) หรือ offset ให้กับ message ทุกตัว (ใน Figure 12-3 , ตัวเลขในกรอบคือ message offset) ทำแบบนี้ได้เพราะ partition (shard) เป็น append-only ดังนั้น message ภายใน partition จึงเรียงลำดับกันอย่างสมบูรณ์ แต่ไม่มีการรับประกันลำดับข้าม partition ที่ต่างกัน

Apache Kafka [ 20 ] และ Amazon Kinesis Streams เป็น log-based message broker ที่ทำงานแบบนี้ Google Cloud Pub/Sub มีสถาปัตยกรรมคล้ายกันแต่เปิดให้ใช้ผ่าน JMS-style API แทนที่จะเป็น log abstraction [ 15 ] แม้ broker เหล่านี้จะเขียน message ทั้งหมดลง disk แต่ก็ได้ throughput หลายล้าน message ต่อวินาทีด้วยการ shard ข้ามหลายเครื่อง และได้ fault tolerance ด้วยการ replicate message [ 21 , 22 ]

Logs compared to traditional messaging (เปรียบเทียบ log กับ messaging แบบดั้งเดิม)

แนวทางแบบ log รองรับ fan-out messaging ได้ง่าย เพราะ consumer หลายตัวอ่าน log ได้อิสระโดยไม่กระทบกัน การอ่าน message ไม่ได้ลบมันออกจาก log ส่วน load balancing ข้ามกลุ่ม consumer broker จะกำหนด shard ทั้งก้อนให้แต่ละ node ใน consumer group แทนที่จะกำหนด message ทีละตัวให้ consumer

แต่ละ client จะ consume ทั้งหมด ของ message ใน shard ที่ได้รับมอบหมาย โดยทั่วไป เมื่อ consumer ได้รับ log shard มาแล้ว จะอ่าน message ใน shard นั้นตามลำดับ แบบ single-thread ง่ายๆ load balancing แบบหยาบนี้มีข้อเสียคือ:

• จำนวน node ที่ช่วยกัน consume topic ได้มากที่สุดเท่ากับจำนวน log shard ใน topic นั้น เพราะ message ใน shard เดียวกันจะถูกส่งไปยัง node เดียวกัน (เราอาจสร้าง load balancing scheme ที่ให้ consumer สองตัวแบ่งงานประมวลผล shard โดยให้ทั้งคู่อ่าน message ครบทุกตัว แต่ตัวหนึ่งสนใจเฉพาะ message ที่ offset เป็นเลขคู่ ส่วนอีกตัวจัดการ offset เลขคี่ หรืออาจกระจายการประมวลผล message ไปยัง thread pool แต่วิธีนั้นทำให้การจัดการ consumer offset ซับซ้อนขึ้น โดยทั่วไปนิยมประมวลผลแบบ single-thread ต่อ shard มากกว่า แล้วเพิ่ม parallelism ด้วยการใช้ shard มากขึ้น)

• ถ้า message ตัวเดียวประมวลผลช้า ก็จะขวางการประมวลผล message ถัดไปใน shard นั้น (รูปแบบหนึ่งของ head-of-line blocking; ดู "Describing Performance" )

ดังนั้น ถ้า message อาจใช้ต้นทุนประมวลผลสูง และคุณต้องการประมวลผลแบบขนานทีละ message และลำดับของ message ไม่สำคัญนัก รูปแบบ JMS/AMQP ของ message broker จะเหมาะกว่า ในทางกลับกัน ในสถานการณ์ที่ message เข้ามาด้วย throughput สูง แต่ละ message ประมวลผลได้เร็ว และลำดับของ message สำคัญ แนวทางแบบ log ทำงานได้ดีมาก [ 23 , 24 ] แต่ความต่างระหว่างสองสถาปัตยกรรมนี้กำลังเลือนลางลง เพราะ messaging system แบบ log อย่าง Kafka ตอนนี้รองรับ consumer group แบบ JMS/AMQP ซึ่งให้ consumer หลายตัวรับ message จาก partition เดียวกันได้ [ 25 , 26 ]

เนื่องจาก sharded log มักรักษาลำดับของ message ได้เฉพาะภายใน shard เดียว message ทั้งหมดที่ต้องเรียงลำดับสอดคล้องกันจึงต้องถูกส่งไปยัง shard เดียวกัน ตัวอย่างเช่น แอปพลิเคชันอาจต้องการให้ event ของผู้ใช้คนใดคนหนึ่งปรากฏตามลำดับที่แน่นอน ซึ่งทำได้โดยเลือก shard ของแต่ละ event จาก user ID ของ event นั้น (พูดอีกอย่างคือให้ user ID เป็น partition key )

Consumer offsets (ตำแหน่ง offset ที่ consumer อ่านถึง)

การ consume shard ตามลำดับทำให้บอกได้ง่ายว่า message ไหนถูกประมวลผลไปแล้ว message ทั้งหมดที่ offset น้อยกว่า consumer offset ปัจจุบันถูกประมวลผลไปแล้ว ส่วน message ที่ offset มากกว่ายังไม่เคยถูกเห็น broker จึงไม่ต้องติดตาม acknowledgment ของทุก message แค่บันทึก consumer offset เป็นระยะก็พอ overhead ของ bookkeeping ที่ลดลง และโอกาสทำ batching กับ pipelining ในแนวทางนี้ ช่วยเพิ่ม throughput ของระบบแบบ log ถ้า consumer ล้มเหลว consumer จะเริ่มใหม่จาก offset ที่บันทึกไว้ล่าสุด ไม่ใช่ offset ล่าสุดที่เคยอ่านถึง ซึ่งอาจทำให้ consumer เห็นบาง message ซ้ำสองครั้ง

ที่จริงแล้ว offset คล้ายกับ log sequence number ที่พบได้ทั่วไปใน replication แบบ single-leader ของฐานข้อมูล ซึ่งเรากล่าวถึงในหัวข้อ "Setting Up New Followers" ใน database replication, log sequence number ช่วยให้ follower กลับมาเชื่อมต่อกับ leader ได้อีกครั้ง หลังจากหลุดการเชื่อมต่อ และทำ replication ต่อได้โดยไม่ข้าม write ใดไป ที่นี่ใช้หลักการเดียวกันเป๊ะ: message broker ทำหน้าที่เหมือน leader database และ consumer ทำหน้าที่เหมือน follower

ถ้า consumer node ล้มเหลว node อื่นใน consumer group จะได้รับมอบหมาย shard ของ consumer ที่ล้มเหลวไป แล้วเริ่ม consume message จาก offset ที่บันทึกไว้ล่าสุด ถ้า consumer เคยประมวลผล message ที่อยู่ถัดจากนั้นไปแล้วแต่ยังไม่ได้บันทึก offset message เหล่านั้นจะถูกประมวลผลซ้ำเป็นครั้งที่สองหลัง restart เราจะพูดถึงวิธีรับมือกับปัญหานี้ต่อไปในบทนี้

Disk space usage (การใช้พื้นที่ disk)

ถ้าเขียนต่อท้าย log ไปเรื่อยๆ ในที่สุด disk ก็จะเต็ม เพื่อคืนพื้นที่ disk จึงแบ่ง log ออกเป็น segment แล้วลบ segment เก่าทิ้งหรือย้ายไปไว้ใน archive storage เป็นระยะ (เราจะพูดถึงวิธีที่ซับซ้อนกว่าในการคืนพื้นที่บน disk ในหัวข้อ "Log compaction" )

นั่นหมายความว่าถ้า consumer ที่ช้าตามทันอัตราของ message ไม่ได้ จนตามหลังมากถึงขั้นที่ consumer offset ชี้ไปยัง segment ที่ถูกลบไปแล้ว consumer ก็จะพลาด message บางส่วนไป โดยพื้นฐานแล้ว log ก็คือ buffer ขนาดจำกัดที่ทิ้ง message เก่าเมื่อเต็ม หรือที่เรียกว่า circular buffer หรือ ring buffer แต่เพราะ buffer นี้อยู่บน disk จึงมีขนาดใหญ่ได้มาก

ลองคำนวณคร่าวๆ ดู ณ ตอนที่เขียนหนังสือเล่มนี้ hard drive ขนาดใหญ่ทั่วไปมีความจุ 20 TB และ sequential write throughput 250 MB/s ถ้าเขียน message ด้วยอัตราเร็วที่สุด จะใช้เวลาประมาณ 22 ชั่วโมงกว่า disk จะเต็ม และต้องเริ่มลบ message ที่เก่าที่สุด นั่นหมายความว่า disk-based log จะ buffer message ได้อย่างน้อย 22 ชั่วโมงเสมอ แม้จะมีหลาย disk และหลายเครื่อง (disk ที่มากขึ้นเพิ่มทั้งพื้นที่ว่างและ total write bandwidth) ในทางปฏิบัติ deployment ส่วนใหญ่ไม่ค่อยใช้ write bandwidth ของ disk จนเต็ม log จึงมักเก็บ buffer ของ message ได้หลายวัน หรือแม้แต่หลายสัปดาห์

ปัจจุบัน log-based message broker หลายตัวเก็บ message ไว้ใน object storage เพื่อเพิ่มความจุ เช่นเดียวกับฐานข้อมูล ดังที่เราเห็นในหัวข้อ "Databases Backed by Object Storage" Message broker อย่าง Apache Kafka และ Redpanda อ่าน message เก่าจาก object storage ซึ่งเป็นส่วนหนึ่งของ tiered storage ส่วนตัวอื่นๆ เช่น WarpStream, Confluent Freight และ Bufstream เก็บข้อมูลทั้งหมดไว้ใน object store นอกจากประหยัดต้นทุนแล้ว สถาปัตยกรรมนี้ยังทำให้การ integrate ข้อมูลง่ายขึ้น: message ใน object storage จะถูกเก็บเป็น Iceberg table ทำให้รัน batch และ data warehouse job บนข้อมูลนั้นได้โดยตรง โดยไม่ต้อง copy ไปไว้ในระบบอื่น

When consumers cannot keep up with producers (เมื่อ consumer ตาม producer ไม่ทัน)

ในตอนต้นของหัวข้อ "Messaging Systems" เราได้กล่าวถึงสามทางเลือกเมื่อ consumer ตามอัตราการส่ง message ของ producer ไม่ทัน: drop message, buffer หรือใช้ backpressure ใน taxonomy นี้ แนวทางแบบ log เป็นรูปแบบหนึ่งของการ buffering ที่ buffer มีขนาดใหญ่แต่คงที่ (ถูกจำกัดด้วยพื้นที่ disk ที่มี)

ถ้า consumer ตามหลังมากจน message ที่ต้องการเก่าเกินกว่าที่เก็บไว้บน disk ก็จะอ่าน message เหล่านั้นไม่ได้ — broker จะทิ้ง message เก่าที่เกินขนาด buffer คุณ monitor ได้ว่า consumer ตามหลัง head ของ log อยู่เท่าไหร่ และตั้ง alert ถ้าตามหลังมากเกินไป เพราะ buffer ใหญ่ จึงมีเวลาพอให้ operator แก้ consumer ที่ช้า ให้ตามทันก่อนที่จะเริ่มพลาด message

แม้ consumer จะตามหลังมากเกินไปจนเริ่มพลาด message ก็กระทบเฉพาะ consumer ตัวนั้น ไม่รบกวนบริการของ consumer ตัวอื่น ข้อนี้เป็นข้อได้เปรียบด้านการ operate ที่สำคัญ คุณลอง consume production log เพื่อพัฒนา, ทดสอบ หรือ debug ได้ โดยไม่ต้องกังวลมากว่าจะกระทบบริการ production เมื่อ consumer ถูกปิดหรือ crash ก็จะเลิกใช้ทรัพยากร — สิ่งเดียวที่เหลืออยู่คือ consumer offset ของมัน

พฤติกรรมนี้ต่างจาก message broker แบบดั้งเดิม ที่ต้องระวังลบ queue ของ consumer ที่ถูกปิดไปแล้ว เพื่อไม่ให้ message สะสมโดยไม่จำเป็นและแย่ง memory จาก consumer ที่ยังทำงานอยู่

Replaying old messages (การ replay message เก่า)

ก่อนหน้านี้เราพูดไปแล้วว่าใน message broker แบบ AMQP และ JMS การประมวลผลและ ack message เป็น destructive operation เพราะทำให้ message ถูกลบบน broker ในทางกลับกัน ใน log-based message broker การ consume message ก็เหมือนการอ่านจากไฟล์: เป็นการดำเนินการแบบ read-only ที่ไม่แก้ไข log

ผลข้างเคียงอย่างเดียวของการประมวลผล นอกเหนือจาก output ของ consumer คือ consumer offset ที่เลื่อนไปข้างหน้า แต่ offset อยู่ในการควบคุมของ consumer จึงปรับได้ง่ายเมื่อจำเป็น — เช่น คุณเริ่ม consumer อีกสำเนาหนึ่งโดยใช้ offset ของเมื่อวาน แล้วเขียน output ไปยังอีกที่หนึ่ง เพื่อประมวลผล message ของวันล่าสุดซ้ำ ทำซ้ำได้กี่ครั้งก็ได้ โดยแก้โค้ดการประมวลผลระหว่างรอบ

ลักษณะนี้ทำให้ log-based messaging คล้าย batch process ที่กล่าวถึงในบทที่แล้ว ซึ่ง derived data ถูกแยกจาก input data อย่างชัดเจนผ่านขั้นตอนการแปลงที่รันซ้ำได้ ทำให้ทดลอง และกู้คืนจากข้อผิดพลาดและบั๊กได้ง่ายขึ้น จึงเป็นเครื่องมือที่ดีสำหรับการ integrate dataflow ภายในองค์กร [ 27 ]

Databases and Streams (ฐานข้อมูลและสตรีม)

เราได้เปรียบเทียบ message broker กับฐานข้อมูลไปบ้างแล้ว แม้โดยดั้งเดิมจะถือว่าเป็นเครื่องมือคนละประเภทกัน แต่เห็นได้ว่า log-based message broker นำแนวคิดจากฐานข้อมูลมาใช้กับ messaging ได้สำเร็จ เรายังทำในทางกลับกันได้ด้วย คือนำแนวคิดจาก messaging และ stream มาใช้กับฐานข้อมูล

แนวทางหนึ่งคือใช้ event stream เป็น system of record ในการเก็บข้อมูล (ดู "Systems of Record and Derived Data" ) นี่คือสิ่งที่เกิดขึ้นใน event sourcing ซึ่งเรากล่าวถึงในหัวข้อ "Event Sourcing and CQRS" แทนที่จะเก็บข้อมูลใน data model ที่ถูกแก้ด้วยการ update และ delete คุณสามารถ model ทุกการเปลี่ยนแปลงสถานะเป็น immutable event และเขียนมันลงใน append-only log materialized view ที่ปรับมาเพื่อการอ่านทั้งหมดจะถูก derive จาก event เหล่านี้ Log-based message broker (ที่ตั้งค่าไม่ให้ลบ event เก่า) เหมาะกับ event sourcing มาก เพราะใช้ append-only storage และแจ้ง consumer เมื่อมี event ใหม่ด้วย latency ต่ำได้

แต่ไม่จำเป็นต้องไปไกลถึงขั้นใช้ event sourcing; แม้กับ data model แบบ mutable ก็ตาม event stream ก็ยังมีประโยชน์กับฐานข้อมูล ที่จริงแล้ว ทุก write ที่เข้าฐานข้อมูลคือ event ที่จับ เก็บ และประมวลผลได้ ความเชื่อมโยงระหว่างฐานข้อมูลกับ stream ลึกกว่าแค่เรื่องการเก็บ log จริงๆ บน disk — ถือว่าเป็นเรื่องพื้นฐานทีเดียว

ตัวอย่างเช่น replication log (ดู "Implementation of Replication Logs" ) คือ stream ของ database write event ที่ leader สร้างขึ้นเมื่อประมวลผล transaction Follower นำ stream ของ write ไป apply กับสำเนาฐานข้อมูลของตัวเอง จึงได้สำเนาข้อมูลชุดเดียวกันที่ถูกต้อง Event ใน replication log บรรยายการเปลี่ยนแปลงข้อมูลที่เกิดขึ้น

เรายังเจอหลักการ state machine replication ในหัวข้อ "Using shared logs" ซึ่งระบุว่า: ถ้าทุก event แทนการ write ไปยังฐานข้อมูล และทุก replica ประมวลผล event ชุดเดียวกันในลำดับเดียวกัน replica ทั้งหมดก็จะจบที่สถานะสุดท้ายเดียวกัน (โดยสมมติว่าการประมวลผล event เป็นการดำเนินการที่ deterministic) นี่ก็เป็นอีกกรณีหนึ่งของ event stream เท่านั้นเอง!

ในส่วนนี้ เราจะดูปัญหาที่เกิดขึ้นใน heterogeneous data system ก่อน แล้วดูว่าแก้ได้อย่างไรโดยนำแนวคิดจาก event stream มาใช้กับฐานข้อมูล

Keeping Systems in Sync (การทำให้ระบบต่างๆ ตรงกัน)

อย่างที่เห็นมาตลอดทั้งเล่ม ไม่มีระบบเดียวที่ตอบโจทย์ทุกความต้องการด้านการเก็บ ค้นหา และประมวลผลข้อมูลได้ ในทางปฏิบัติ แอปพลิเคชันที่ไม่ธรรมดาส่วนใหญ่ต้องผสมหลายเทคโนโลยีเข้าด้วยกัน เพื่อให้ตรงตามความต้องการ — เช่น ใช้ OLTP database รับ user request, ใช้ cache เร่ง request ที่พบบ่อย, ใช้ full-text index จัดการ search query และใช้ data warehouse สำหรับ analytics แต่ละระบบมีสำเนาข้อมูลของตัวเอง เก็บในรูปแบบที่ปรับมาให้เหมาะกับงานของตัวเอง

เมื่อข้อมูลเดียวกันหรือข้อมูลที่เกี่ยวข้องกันอยู่ในหลายที่ ก็ต้องทำให้ข้อมูลเหล่านั้นตรงกัน ถ้า item หนึ่งถูก update ในฐานข้อมูล ก็ต้อง update ใน cache, search index และ data warehouse ด้วย สำหรับ data warehouse การ sync นี้มักทำโดย ETL process (ดู "Data Warehousing" ) ซึ่งมักทำโดยคัดลอกฐานข้อมูลทั้งก้อน, แปลงข้อมูล แล้ว bulk-load เข้า data warehouse — พูดอีกอย่างคือเป็น batch process ในทำนองเดียวกัน เราเห็นในหัวข้อ "Batch Use Cases" ว่า search index, recommendation system และ derived data system อื่นๆ อาจสร้างได้ด้วย batch process

ถ้าการ dump ฐานข้อมูลทั้งก้อนเป็นระยะช้าเกินไป ทางเลือกที่บางครั้งใช้คือ dual writes (การเขียนสองทาง) ซึ่งโค้ดแอปพลิเคชันจะเขียนไปยังแต่ละระบบโดยตรงเมื่อข้อมูลเปลี่ยนแปลง — เช่น เขียนลงฐานข้อมูลก่อน แล้ว update search index แล้ว invalidate cache entry (หรืออาจทำ write เหล่านั้นพร้อมกันด้วยซ้ำ)

แต่ dual writes มีปัญหาร้ายแรง หนึ่งในนั้นคือ race condition ดังที่แสดงใน Figure 12-4 ในตัวอย่างนี้ client สองตัวต้องการ update item X พร้อมกัน Client 1 ต้องการตั้งค่าเป็น A และ client 2 ต้องการตั้งค่าเป็น B ทั้งสอง client เขียนค่าใหม่ลงฐานข้อมูลก่อน แล้วจึงเขียนไปยัง search index เพราะจังหวะที่ไม่เป็นใจ request จึงถูกสลับลำดับกัน ฐานข้อมูลเห็น write จาก client 1 ที่ตั้งค่าเป็น A ก่อน แล้ว write จาก client 2 ที่ตั้งค่าเป็น B ดังนั้นค่าสุดท้ายในฐานข้อมูลคือ B Search index เห็น write จาก client 2 ก่อน แล้วตามด้วย client 1 ดังนั้นค่าสุดท้ายใน search index คือ A ตอนนี้ทั้งสองระบบไม่ตรงกันถาวร ทั้งที่ไม่มี error เกิดขึ้นเลย

Figure 12-4. ในฐานข้อมูล X ถูกตั้งเป็น A ก่อนแล้วจึงเป็น B ส่วนที่ search index write มาถึงในลำดับตรงข้าม

ถ้าไม่มีกลไกตรวจจับ concurrency เพิ่มเติม อย่าง version vector ที่เราพูดถึงในหัวข้อ "Detecting Concurrent Writes" คุณจะไม่รู้ด้วยซ้ำว่าเกิด concurrent write ค่าหนึ่งแค่เขียนทับอีกค่าหนึ่งไปเงียบๆ

อีกปัญหาหนึ่งของ dual writes คือ write หนึ่งอาจล้มเหลวในขณะที่อีก write สำเร็จ นี่เป็นปัญหาด้าน fault tolerance มากกว่า concurrency แต่ผลก็คือทั้งสองระบบไม่ตรงกันเหมือนกัน การทำให้แน่ใจว่าทั้งสองฝั่งสำเร็จทั้งคู่หรือล้มเหลวทั้งคู่ คือ atomic commit problem ซึ่งแก้ได้แต่ต้นทุนสูง (ดู "Two-Phase Commit" )

ถ้ามีฐานข้อมูลที่ replicate เพียงชุดเดียวและมี leader ตัวเดียว leader ตัวนั้นเป็นผู้กำหนดลำดับของ write แนวทาง state machine replication จึงใช้ได้ระหว่าง replica ของฐานข้อมูล แต่ใน Figure 12-4 ไม่มี leader ตัวเดียว ฐานข้อมูลอาจมี leader ของตัวเอง และ search index ก็อาจมี leader ของตัวเอง แต่ไม่มีฝ่ายใดตามอีกฝ่าย จึงเกิด conflict ได้ (ดู "Multi-Leader Replication" )

ถ้ามี leader แค่ตัวเดียวจริงๆ สถานการณ์จะดีกว่านี้ — เช่น ให้ฐานข้อมูลเป็น leader — และถ้าเราทำให้ search index เป็น follower ของฐานข้อมูลได้ แต่ในทางปฏิบัติทำได้หรือไม่?

Change Data Capture (การจับภาพการเปลี่ยนแปลงข้อมูล)

ปัญหาของ replication log ในฐานข้อมูลส่วนใหญ่คือ ถูกมองกันมานานว่าเป็นรายละเอียดการ implement ภายในของฐานข้อมูล ไม่ใช่ public API Client ควร query ฐานข้อมูลผ่าน data model และ query language ไม่ใช่ parse replication log แล้วพยายามดึงข้อมูลออกมา

เป็นเวลาหลายสิบปีที่ฐานข้อมูลจำนวนมากไม่มีวิธีที่มีเอกสารรองรับ ในการดึง log ของการเปลี่ยนแปลงที่เขียนลงไป ซึ่งทำให้ยากที่จะนำการเปลี่ยนแปลงทั้งหมดในฐานข้อมูลไป replicate ยังเทคโนโลยีเก็บข้อมูลอื่น เช่น search index, cache หรือ data warehouse

ช่วงหลังมานี้ มีความสนใจเพิ่มขึ้นใน change data capture (CDC) ซึ่งเป็นกระบวนการสังเกตการเปลี่ยนแปลงข้อมูลทั้งหมดที่เขียนลงฐานข้อมูล และดึงออกมาในรูปแบบที่ replicate ไปยังระบบอื่นได้ [ 28 ] CDC น่าสนใจเป็นพิเศษถ้าการเปลี่ยนแปลงถูกเปิดให้ใช้เป็น stream ทันทีที่ถูกเขียน

ตัวอย่างเช่น คุณสามารถจับการเปลี่ยนแปลงในฐานข้อมูล แล้ว apply การเปลี่ยนแปลงเดียวกันกับ search index อย่างต่อเนื่อง ถ้า apply log ของการเปลี่ยนแปลงตามลำดับเดียวกัน ก็คาดหวังได้ว่าข้อมูลใน search index จะตรงกับในฐานข้อมูล Search index และ derived data system อื่นๆ ก็เป็นแค่ consumer ของ change stream

Figure 12-5 แสดงให้เห็นว่า CDC แก้ปัญหา concurrency ของ Figure 12-4 ได้อย่างไร แม้ request สองรายการที่ตั้งค่า X เป็น A และ B ตามลำดับจะมาถึงฐานข้อมูลพร้อมกัน ฐานข้อมูลจะเป็นผู้ตัดสินลำดับการ execute แล้วเขียนลง replication log ตามลำดับนั้น Search index จะดึงไปและ apply ตามลำดับเดียวกัน ถ้าต้องการข้อมูลในระบบอื่น เช่น data warehouse ก็แค่เพิ่มมันเป็น consumer อีกตัวของ CDC event stream

Figure 12-5. นำการเปลี่ยนแปลงที่ commit ลงฐานข้อมูลแล้วส่งต่อไปยัง downstream system ตามลำดับเดียวกัน

Implementing CDC (การนำ CDC ไปปฏิบัติ)

เราเรียก log consumer เหล่านี้ว่า derived data systems ดังที่กล่าวไว้ในหัวข้อ "Systems of Record and Derived Data" ข้อมูลที่เก็บใน search index และ data warehouse ก็เป็นแค่อีกมุมมองหนึ่งของข้อมูลใน system of record CDC เป็นกลไกที่รับประกันว่าทุกการเปลี่ยนแปลงที่เกิดกับ system of record จะไปปรากฏใน derived data system ด้วย เพื่อให้ derived system มีสำเนาข้อมูลที่ถูกต้อง

โดยพื้นฐานแล้ว CDC ทำให้ฐานข้อมูลหนึ่งตัวเป็น leader (ตัวที่ถูกดึงการเปลี่ยนแปลงออกมา) และเปลี่ยนตัวอื่นๆ ให้เป็น follower Log-based message broker เหมาะมากสำหรับส่ง change event จาก source database ไปยัง derived system เพราะรักษาลำดับของ message ไว้ (เลี่ยงปัญหา reordering ใน Figure 12-2 )

ใช้ logical replication log implement CDC ได้ (ดู "Logical (row-based) log replication" ) แม้จะมีความท้าทาย เช่น การจัดการ schema change และการ model การ update ให้ถูกต้อง โปรเจกต์โอเพนซอร์ส Debezium แก้ปัญหาเหล่านี้ โปรเจกต์นี้มี source connectors สำหรับ MySQL, PostgreSQL, Oracle, SQL Server, Db2, Cassandra, และฐานข้อมูลอื่นๆ อีกมากมาย Connector เหล่านี้เชื่อมต่อกับ database replication log แล้วแสดงการเปลี่ยนแปลงในรูปแบบ event schema มาตรฐาน จากนั้น message จะถูกแปลงและเขียนไปยัง downstream database ได้ Kafka Connect framework ก็มี CDC connector สำหรับฐานข้อมูลต่างๆ เช่นกัน Maxwell ทำสิ่งที่คล้ายกันสำหรับ MySQL โดย parse binlog [ 29 ], GoldenGate มีความสามารถคล้ายกันสำหรับ Oracle และ pgcapture ทำเช่นเดียวกันสำหรับ PostgreSQL

เช่นเดียวกับ message broker, CDC มักเป็น asynchronous: ฐานข้อมูล system-of-record ไม่รอให้ consumer apply การเปลี่ยนแปลงเสร็จก่อนจึง commit การออกแบบนี้มีข้อดีด้านการ operate คือการเพิ่ม consumer ที่ช้า ไม่กระทบ system of record มากนัก แต่ก็มีข้อเสียคือปัญหา replication lag ทุกอย่างจะเกิดขึ้นด้วย (ดู "Problems with Replication Lag" )

Initial snapshot (ภาพรวมเริ่มต้น)

ถ้ามี log ของทุกการเปลี่ยนแปลงที่เคยเกิดขึ้นในฐานข้อมูล ก็สร้างสถานะทั้งหมดของฐานข้อมูลขึ้นใหม่ได้ด้วยการ replay log แต่ในหลายกรณี การเก็บทุกการเปลี่ยนแปลงไว้ตลอดไป ต้องใช้พื้นที่ disk มากเกินไป และการ replay ก็ใช้เวลานานเกินไป log จึงต้องถูกตัดทอน (truncate)

การสร้าง full-text index ใหม่ ตัวอย่างเช่น ต้องใช้สำเนาของฐานข้อมูลทั้งก้อน การ apply แค่ log ของการเปลี่ยนแปลงล่าสุดไม่เพียงพอ เพราะจะขาด item ที่ไม่ได้ถูก update เมื่อเร็วๆ นี้ ดังนั้น ถ้าไม่มีประวัติ log ครบทั้งหมด ก็ต้องเริ่มจาก consistent snapshot ดังที่ได้กล่าวถึงในหัวข้อ "Setting Up New Followers"

snapshot ของฐานข้อมูลต้องตรงกับตำแหน่งหรือ offset ที่ทราบแน่ชัดใน change log เพื่อให้รู้ว่าต้องเริ่ม apply การเปลี่ยนแปลงจากจุดไหนหลังจากประมวลผล snapshot เสร็จแล้ว เครื่องมือ CDC บางตัวมีความสามารถทำ snapshot นี้ในตัว ส่วนบางตัวปล่อยให้ทำเองด้วยมือ Debezium ใช้อัลกอริทึม DBLog watermarking ของ Netflix เพื่อทำ incremental snapshot [ 30 , 31 ]

Log compaction (การบีบอัด Log)

ถ้าเก็บประวัติ log ได้เพียงจำนวนจำกัด ก็ต้องทำ snapshot ทุกครั้งที่ต้องการเพิ่ม derived data system ใหม่ แต่ log compaction เป็นทางเลือกที่ดี

เราเคยพูดถึง log compaction ก่อนหน้านี้ในหัวข้อ "Log-Structured Storage" ในบริบทของ log-structured storage engine (ดู Figure 4-3 สำหรับตัวอย่าง) หลักการนั้นง่าย: storage engine จะค้นหา log record ที่มี key เดียวกันเป็นระยะ ทิ้งตัวซ้ำ และเก็บเฉพาะ update ล่าสุดของแต่ละ key ซึ่งทำให้ log segment เล็กลงมาก จึงอาจรวม segment เข้าด้วยกันไปพร้อมกันในกระบวนการ compaction ดังแสดงใน Figure 12-6 กระบวนการนี้ทำงานเบื้องหลัง

Figure 12-6. ใน log ของ key-value pair นี้ key คือ ID ของวิดีโอแมว (mew, purr, scratch หรือ yawn) และ value คือจำนวนครั้งที่ถูกเล่น; log compaction จะเก็บเฉพาะค่าล่าสุดของแต่ละ key

ใน log-structured storage engine การ update ด้วยค่า null พิเศษ (เรียกว่า tombstone ) บ่งชี้ว่า key ถูกลบ และทำให้ key นั้นถูกเอาออกระหว่าง log compaction แต่ตราบใดที่ key ยังไม่ถูกเขียนทับหรือลบ ก็จะอยู่ใน log ตลอดไป พื้นที่ disk ที่ compacted log ต้องใช้ขึ้นอยู่กับเนื้อหาปัจจุบันของฐานข้อมูลเท่านั้น ไม่ขึ้นกับจำนวน write ที่เคยเกิดขึ้นในฐานข้อมูล ถ้า key เดียวกันถูกเขียนทับบ่อย ค่าก่อนหน้าจะถูก garbage-collect ในที่สุด และจะเก็บเฉพาะค่าล่าสุดไว้

แนวคิดเดียวกันนี้ใช้ได้กับ log-based message broker และ CDC ถ้าตั้งระบบ CDC ให้ทุกการเปลี่ยนแปลงมี primary key และทุก update ของ key หนึ่งแทนที่ค่าก่อนหน้าของ key นั้น ก็เพียงพอแล้วที่จะเก็บแค่ write ล่าสุดของ key นั้น

ตอนนี้ เมื่อใดก็ตามที่ต้องการสร้าง derived data system ใหม่ เช่น search index ก็เริ่ม consumer ใหม่จาก offset 0 ของ topic ที่ทำ log compaction ไว้ได้ แล้วสแกน message ทั้งหมดใน log ตามลำดับ Log รับประกันว่าจะมีค่าล่าสุดของทุก key ในฐานข้อมูล (และอาจมีค่าที่เก่ากว่าบ้าง) พูดอีกอย่างคือ ใช้มันเพื่อได้สำเนาเนื้อหาฐานข้อมูลทั้งหมดได้ โดยไม่ต้อง snapshot CDC source database ซ้ำอีก

Apache Kafka รองรับฟีเจอร์ log compaction นี้ ดังที่เราจะเห็นต่อไปในบทนี้ ฟีเจอร์นี้ทำให้ใช้ message broker เป็นที่เก็บข้อมูลแบบ durable ได้ ไม่ใช่แค่ส่ง message ชั่วคราว

API support for change streams (การสนับสนุน API สำหรับ Change Streams)

ปัจจุบันฐานข้อมูลยอดนิยมส่วนใหญ่เปิด change stream ให้ใช้เป็น first-class interface แทนที่จะเป็นความพยายามทำ CDC แบบต่อเติมและ reverse-engineer ในอดีต relational database อย่าง MySQL และ PostgreSQL มักส่งการเปลี่ยนแปลงผ่าน replication log เดียวกับที่ใช้กับ replica ของตัวเอง cloud vendor ส่วนใหญ่ก็มีโซลูชัน CDC สำหรับผลิตภัณฑ์ของตนเช่นกัน — เช่น Datastream ให้เข้าถึงข้อมูลแบบ streaming สำหรับ relational database และ data warehouse ของ Google Cloud

แม้แต่ฐานข้อมูลแบบ eventually consistent และ quorum-based อย่าง Cassandra ตอนนี้ก็รองรับ CDC แล้ว ดังที่เราเห็นในหัวข้อ "Implementing Linearizable Systems" client ต้อง persist write ไปยัง majority ของ node ก่อนจึงจะถือว่า visible การรองรับ CDC สำหรับ quorum write จึงเป็นเรื่องท้าทาย เพราะไม่มี single source of truth ให้ subscribe ข้อมูลจะ visible หรือไม่ ขึ้นอยู่กับ consistency preference ของผู้อ่านแต่ละราย Cassandra เลี่ยงปัญหานี้โดยเปิดให้ใช้ raw log segment ของแต่ละ node แทนที่จะให้ stream เดียวของ mutation ระบบที่ต้องการ consume ข้อมูลต้องอ่าน raw log segment ของแต่ละ node แล้วตัดสินใจว่าจะรวมเป็น stream เดียวอย่างไรดีที่สุด (คล้ายสิ่งที่ quorum reader ทำ) [ 32 ]

Kafka Connect [ 33 ] เชื่อมเครื่องมือ CDC สำหรับระบบฐานข้อมูลหลากหลายเข้ากับ Kafka เมื่อ stream ของ change event อยู่ใน Kafka แล้ว ก็ใช้ update derived data system เช่น search index ได้ รวมถึงป้อนเข้าระบบ stream processing ดังที่จะกล่าวต่อไปในบทนี้

CDC versus event sourcing (CDC เปรียบเทียบกับ Event Sourcing)

แล้ว CDC ต่างจาก event sourcing อย่างไร? คล้ายกับ CDC, event sourcing คือการเก็บทุกการเปลี่ยนแปลง ของ application state ไว้เป็น log ของ change event ความต่างที่ใหญ่ที่สุดคือ event sourcing ใช้แนวคิดนี้ที่ระดับ abstraction ต่างออกไป:

• ใน CDC แอปพลิเคชันใช้ฐานข้อมูลแบบ mutable คือ update และ delete record ได้ตามใจ log ของการเปลี่ยนแปลงถูกดึงออกมาจากฐานข้อมูลที่ระดับต่ำ (เช่น โดย parse replication log) ซึ่งทำให้แน่ใจว่าลำดับของ write ที่ดึงออกมาจากฐานข้อมูล ตรงกับลำดับที่ถูกเขียนจริง จึงเลี่ยง race condition ใน Figure 12-4

• ใน event sourcing logic ของแอปพลิเคชันถูกสร้างขึ้นโดยตั้งใจ บนพื้นฐานของ immutable event ที่เขียนลง event log ในกรณีนี้ event store เป็น append-only และไม่ส่งเสริมหรือห้ามการ update หรือ delete event Event ถูกออกแบบมาให้สะท้อนสิ่งที่เกิดขึ้นในระดับแอปพลิเคชัน ไม่ใช่การเปลี่ยนแปลง state ระดับต่ำ

แบบไหนดีกว่าขึ้นอยู่กับสถานการณ์ของคุณ การนำ event sourcing มาใช้เป็นการเปลี่ยนแปลงครั้งใหญ่สำหรับแอปพลิเคชัน ที่ยังไม่ได้ทำมัน และมีข้อดีข้อเสียหลายอย่าง ซึ่งเราได้พูดถึงในหัวข้อ "Event Sourcing and CQRS" ในทางกลับกัน เราเพิ่ม CDC เข้ากับฐานข้อมูลที่มีอยู่แล้วได้ โดยแก้ไขแค่เล็กน้อย แอปพลิเคชันที่เขียนไปยังฐานข้อมูลอาจไม่รู้ด้วยซ้ำว่า CDC กำลังเกิดขึ้น

Change Data Capture and Database Schemas (CDC กับ schema ของฐานข้อมูล)

แม้ CDC จะดูเหมือนนำมาใช้ได้ง่ายกว่า event sourcing แต่ก็มีความท้าทายของตัวเองเหมือนกัน ในสถาปัตยกรรม microservices ฐานข้อมูลมักถูกเข้าถึงจาก service เดียวเท่านั้น Service อื่นโต้ตอบกับมันผ่าน public API ของ service นั้น และปกติจะไม่เข้าถึงฐานข้อมูลโดยตรง ฐานข้อมูลจึงเป็นรายละเอียดการ implement ภายในของ service ทำให้นักพัฒนาเปลี่ยน schema ได้โดยไม่กระทบ public API

แต่ระบบ CDC มักใช้ schema ของ upstream database เมื่อ replicate ข้อมูล ซึ่งทำให้ schema เหล่านี้กลายเป็น public API ที่ต้องจัดการไม่ต่างจาก public API ของ service การลบ column ในตารางฐานข้อมูลจะทำให้ downstream consumer ที่พึ่งพา field นั้นพัง ความท้าทายแบบนี้มีอยู่เสมอกับ data pipeline แต่ปกติกระทบเฉพาะ ETL ของ data warehouse เนื่องจาก CDC มักถูก implement เป็น data stream service production อื่นๆ จึงอาจเป็น consumer ด้วย การทำให้ consumer เหล่านี้พังอาจทำให้เกิด outage ที่กระทบลูกค้า [ 34 ] มักใช้ data contract เพื่อป้องกันความเสียหายเหล่านี้

วิธีที่พบบ่อยในการแยก internal schema ออกจาก external schema คือใช้ outbox pattern Outbox คือตารางที่มี schema ของตัวเอง ซึ่งเปิดให้ระบบ CDC ใช้แทนที่จะเปิด domain model ภายในของฐานข้อมูล [ 35 , 36 ] นักพัฒนาจึงปรับเปลี่ยน internal schema ได้ตามที่เห็นสมควร โดยไม่ต้องแตะ outbox table ซึ่งอาจดูเหมือน dual write — และก็เป็น dual write จริงๆ แต่ outbox เลี่ยงความท้าทายที่เราพูดถึงในหัวข้อ "Keeping Systems in Sync" ด้วยการเก็บ write ทั้งสองไว้ในระบบเดียวกัน (ฐานข้อมูล) การออกแบบนี้ทำให้ write ทั้งสองอยู่ใน transaction เดียวกันได้

แต่ outbox ก็มี trade-off อยู่บ้าง นักพัฒนายังต้องดูแลการแปลงระหว่าง internal schema กับ outbox schema ซึ่งอาจเป็นเรื่องท้าทาย Outbox ยังเพิ่มปริมาณข้อมูลที่ฐานข้อมูลต้องเขียนลง storage พื้นฐาน ซึ่งอาจทำให้เกิดปัญหาด้านประสิทธิภาพ

เช่นเดียวกับ CDC การ replay event log ช่วยให้สร้างสถานะปัจจุบันของระบบขึ้นใหม่ได้ แต่ log compaction ต้องจัดการต่างออกไป:

• CDC event ของการ update record มักบรรจุ record เวอร์ชันใหม่ทั้งหมด ค่าปัจจุบันของ primary key จึงถูกกำหนดจาก event ล่าสุดของ primary key นั้นทั้งหมด และ log compaction ทิ้ง event ก่อนหน้าของ key เดียวกันได้

• ในทางกลับกัน ใน event sourcing event ถูก model ที่ระดับสูงกว่า Event มักแสดงเจตนาของการกระทำของผู้ใช้ ไม่ใช่กลไกการ update state ที่เกิดขึ้นเป็นผลจากการกระทำนั้น ในกรณีนี้ event หลังๆ มักไม่แทนที่ event ก่อนหน้า คุณจึงต้องมีประวัติ event ทั้งหมดเพื่อสร้าง state สุดท้าย จึงทำ log compaction แบบเดียวกันไม่ได้

แอปพลิเคชันที่ใช้ event sourcing มักมีกลไกเก็บ snapshot ของสถานะปัจจุบันที่ derive จาก log ของ event จะได้ไม่ต้องประมวลผล log ทั้งหมดซ้ำบ่อยๆ แต่นี่เป็นแค่การ optimize ประสิทธิภาพเพื่อเร่งการอ่านและการกู้คืนจาก crash เจตนาคือให้ระบบเก็บ raw event ทั้งหมดไว้ได้ตลอดไป และประมวลผล event log ทั้งหมดซ้ำได้เมื่อจำเป็น เราจะพูดถึงสมมติฐานนี้ในหัวข้อ "Limitations of immutability"

State, Streams, and Immutability (สถานะ, สตรีม, และความไม่เปลี่ยนรูป)

เราได้เห็นใน บทที่ 11 ว่า batch processing ได้ประโยชน์จาก immutability ของ input file คุณจึงรัน experimental processing job บน input file ที่มีอยู่ได้ โดยไม่ต้องกลัวว่าจะทำให้เสียหาย หลักการ immutability นี้เองที่ทำให้ event sourcing และ CDC ทรงพลังมาก

โดยปกติเรามองว่าฐานข้อมูลเก็บ state ปัจจุบันของแอปพลิเคชัน การแทนข้อมูลแบบนี้ถูกปรับมาเพื่อการอ่าน และโดยปกติสะดวกที่สุดสำหรับการตอบ query ธรรมชาติของ state คือเปลี่ยนแปลงได้ ฐานข้อมูลจึงรองรับการ update และ delete ข้อมูล รวมถึงการ insert แล้วสิ่งนี้เข้ากับ immutability ได้อย่างไร?

เมื่อใดก็ตามที่มี state ที่เปลี่ยนแปลง state นั้นเป็นผลลัพธ์ของ event ที่ทำให้มันเปลี่ยนไปเรื่อยๆ ตามกาลเวลา ตัวอย่างเช่น รายการที่นั่งที่ว่างอยู่ตอนนี้คือผลลัพธ์ของการจองที่คุณประมวลผลไปแล้ว ยอดเงินในบัญชีปัจจุบันคือผลลัพธ์ของ credit และ debit ในบัญชี และกราฟ response time ของ web server ของคุณคือผลรวม (aggregation) ของ response time ของแต่ละ web request ที่เกิดขึ้นทั้งหมด

ไม่ว่า state จะเปลี่ยนไปอย่างไร ก็มีลำดับของ event ที่ทำให้เกิดการเปลี่ยนแปลงเหล่านั้นเสมอ แม้สิ่งต่างๆ จะถูกทำแล้วถูกยกเลิก ความจริงที่ว่า event เหล่านั้นเคยเกิดขึ้นก็ยังคงอยู่ แนวคิดสำคัญคือ mutable state กับ append-only log ของ immutable event ไม่ได้ขัดแย้งกัน; ทั้งสองเป็นสองด้านของเหรียญเดียวกัน log ของการเปลี่ยนแปลงทั้งหมด หรือ changelog แสดงถึงวิวัฒนาการของ state ตามเวลา

ถ้าคุณถนัดคณิตศาสตร์ คุณอาจบอกว่า application state คือสิ่งที่ได้ เมื่อ integrate event stream ตามเวลา และ change stream คือสิ่งที่ได้เมื่อ differentiate state ตามเวลา ดังแสดงใน Figure 12-7 [ 37 , 38 ] การเปรียบเทียบนี้มีข้อจำกัด (เช่น อนุพันธ์อันดับสองของ state ดูเหมือนจะไม่มีความหมาย) แต่ก็เป็นจุดเริ่มต้นที่เป็นประโยชน์สำหรับการคิดเรื่องข้อมูล

Figure 12-7. ความสัมพันธ์ระหว่าง application state ปัจจุบันกับ event stream

ถ้าเก็บ changelog แบบ durable ก็แค่ทำให้ state สร้างซ้ำได้ (reproducible) ถ้าถือว่า log ของ event คือ system of record และ mutable state ใดๆ เป็นสิ่งที่ derive มาจากมัน ก็จะวิเคราะห์การไหลของข้อมูลผ่านระบบได้ง่ายขึ้น ดังที่ Jim Gray และ Andreas Reuter กล่าวไว้ในปี 1992 [ 39 ]:

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

Log compaction เป็นวิธีหนึ่งที่เชื่อมความต่างระหว่าง log กับ state ของฐานข้อมูล การ compact จะเก็บเฉพาะ version ล่าสุดของแต่ละ record และทิ้ง version ที่ถูกเขียนทับ

Advantages of immutable events (ข้อดีของ event ที่ immutable)

Immutability ในฐานข้อมูลเป็นแนวคิดเก่า เช่น นักบัญชีใช้ immutability มาหลายศตวรรษแล้วในการทำบัญชีการเงิน เมื่อเกิดธุรกรรมขึ้น จะถูกบันทึกลงใน append-only ledger (สมุดบัญชี) ซึ่งโดยพื้นฐานคือ log ของ event ที่บันทึกเงิน, สินค้า หรือบริการที่เปลี่ยนมือ บัญชีสรุปอย่างงบกำไรขาดทุนหรืองบดุล ถูก derive จากธุรกรรมใน ledger โดยการรวมยอดเข้าด้วยกัน [ 40 ]

ถ้าเกิดข้อผิดพลาดขึ้น นักบัญชีจะไม่ลบหรือแก้ธุรกรรมที่ผิดใน ledger แต่จะเพิ่มธุรกรรมใหม่อีกรายการเพื่อชดเชยข้อผิดพลาดแทน — เช่น การคืนเงินสำหรับการเรียกเก็บเงินที่ผิดพลาด ธุรกรรมที่ผิดจะยังคงอยู่ใน ledger ตลอดไป เพราะอาจสำคัญต่อการตรวจสอบบัญชี (audit) ถ้าตัวเลขที่ผิดซึ่ง derive จาก ledger ที่ผิดถูกเผยแพร่ไปแล้ว ตัวเลขของงวดบัญชีถัดไปจะรวมการแก้ไขนั้นเข้าไปด้วย กระบวนการนี้เป็นเรื่องปกติอย่างยิ่งในวงการบัญชี [ 41 ]

แม้ auditability แบบนี้จะสำคัญเป็นพิเศษในระบบการเงิน แต่ก็เป็นประโยชน์กับระบบอื่นอีกมากที่ไม่ได้อยู่ภายใต้กฎระเบียบเข้มงวดขนาดนั้น ถ้าคุณเผลอ deploy โค้ดที่มีบั๊กซึ่งเขียนข้อมูลเสียลงฐานข้อมูล การกู้คืนจะยากกว่ามากถ้าโค้ดเขียนทับข้อมูลเดิมแบบทำลายทิ้งได้ แต่ถ้าเป็น append-only log ของ immutable event การวินิจฉัยว่าเกิดอะไรขึ้นและการกู้คืนจากปัญหาง่ายกว่ามาก ในทำนองเดียวกัน ฝ่ายบริการลูกค้าก็ใช้ audit log ได้ เพื่อวินิจฉัยคำขอและข้อร้องเรียนของลูกค้า

Immutable event ยังเก็บข้อมูลได้มากกว่าแค่ state ปัจจุบัน เช่น บนเว็บไซต์ช้อปปิ้ง ลูกค้าอาจเพิ่มสินค้าในตะกร้าแล้วลบออกอีกครั้ง แม้ event ที่สองจะหักล้าง event แรกในมุมมองของการ fulfill คำสั่งซื้อ แต่ในมุมของ analytics การรู้ว่าลูกค้าเคยพิจารณาสินค้าชิ้นหนึ่ง ก่อนจะตัดสินใจไม่ซื้อ อาจมีประโยชน์ บางทีเขาอาจซื้อในอนาคต หรือบางทีอาจเจอสินค้าทดแทน ข้อมูลนี้ถูกบันทึกไว้ใน event log แต่จะหายไปในฐานข้อมูลที่ลบสินค้าทิ้งเมื่อถูกเอาออกจากตะกร้า

Deriving several views from the same event log (การ derive หลาย view จาก event log เดียวกัน)

โดยการแยก mutable state ออกจาก immutable event log คุณ derive การแทนข้อมูลแบบ read-oriented ได้หลายแบบ จาก log ของ event เดียวกัน ซึ่งทำงานเหมือนกับการมี consumer หลายตัวของ stream (Figure 12-5 ) — เช่น analytical database อย่าง Druid ingest จาก Kafka โดยตรงด้วยแนวทางนี้ และ Kafka Connect sink ก็ export ข้อมูลจาก Kafka ไปยังฐานข้อมูลและ index ต่างๆ ได้ [ 33 ]

การมีขั้นตอนแปลงที่ชัดเจนจาก event log ไปยังฐานข้อมูล ทำให้พัฒนาแอปพลิเคชันต่อไปตามกาลเวลาได้ง่ายขึ้น ถ้าต้องการเพิ่มฟีเจอร์ใหม่ที่นำเสนอข้อมูลเดิมของคุณในรูปแบบใหม่ ก็ใช้ event log สร้าง read-optimized view แยกต่างหากสำหรับฟีเจอร์ใหม่นั้นได้ แล้วรันควบคู่กับระบบเดิมโดยไม่ต้องแก้ไขระบบเดิม การรันระบบเก่าและใหม่เคียงข้างกันมักง่ายกว่า การทำ schema migration ที่ซับซ้อนบนระบบเดิม เมื่อ reader ย้ายไปใช้ระบบใหม่หมดแล้วและไม่ต้องใช้ระบบเก่าอีก ก็ปิดระบบเก่าและคืนทรัพยากรได้เลย [ 42 , 43 ]

เราเคยเจอแนวคิดนี้มาแล้ว คือเขียนข้อมูลในรูปแบบที่ปรับมาเพื่อการเขียน แล้วแปลงเป็น read-optimized representation ต่างๆ ตามต้องการ ในหัวข้อ "Event Sourcing and CQRS" กระบวนการนี้ไม่จำเป็นต้องใช้ event sourcing เสมอไป คุณสร้าง materialized view หลายตัว จาก stream ของ CDC event ได้เหมือนกัน [ 44 ]

แนวทางดั้งเดิมในการออกแบบฐานข้อมูลและ schema ตั้งอยู่บนความเข้าใจผิดว่าข้อมูลต้องถูกเขียนในรูปแบบเดียวกับที่จะถูก query การถกเถียงเรื่อง normalization กับ denormalization (ดู "Normalization, Denormalization, and Joins" ) แทบไม่เกี่ยวข้องอีกต่อไป ถ้าคุณแปลงข้อมูลจาก write-optimized event log ไปเป็น read-optimized application state ได้ การ denormalize ข้อมูลใน read-optimized view จึงสมเหตุสมผลอย่างยิ่ง เพราะกระบวนการแปลงมีกลไกที่ทำให้ข้อมูลสอดคล้องกับ event log

ในหัวข้อ "Case Study: Social Network Home Timelines" เราได้พูดถึง home timeline ของโซเชียลเน็ตเวิร์ก ซึ่งเป็น cache ของโพสต์ล่าสุดจากคนที่ผู้ใช้คนหนึ่งติดตามอยู่ (เหมือนกล่องจดหมาย) นี่เป็นอีกตัวอย่างของ read-optimized state: home timeline ถูก denormalize อย่างหนัก เพราะโพสต์ของคุณถูกทำซ้ำใน timeline ของทุกคนที่ติดตามคุณ แต่ fan-out service จะทำให้ state ที่ซ้ำกันนี้ sync กับโพสต์ใหม่ และความสัมพันธ์การติดตามใหม่ ซึ่งทำให้ข้อมูลที่ซ้ำกันยังจัดการได้ (manageable)

Concurrency control (การควบคุมการทำงานพร้อมกัน)

ข้อเสียที่ใหญ่ที่สุดของ CQRS คือ consumer ของ event log มักเป็น asynchronous ผู้ใช้จึงอาจ write ลง log แล้วอ่านจาก derived view และพบว่า write ของตัวเองยังไม่ปรากฏใน view เราได้พูดถึงปัญหานี้และแนวทางแก้ที่เป็นไปได้ในหัวข้อ "Reading your own writes"

วิธีแก้วิธีหนึ่งคือ update read view แบบ synchronous ไปพร้อมกับการต่อท้าย event ลง log ซึ่งต้องใช้ distributed transaction ข้าม event log และ derived view หรือต้องมีวิธีรอจนกว่า event จะปรากฏใน view ทั้งสองวิธีมักใช้ไม่ได้จริงในทางปฏิบัติ view จึงมักถูก update แบบ asynchronous

ในทางกลับกัน การ derive state ปัจจุบันจาก event log ยังช่วยให้บางแง่มุมของ concurrency control ง่ายขึ้นด้วย ความต้องการ multi-object transaction (ดู "Single-Object and Multi-Object Operations" ) ส่วนใหญ่เกิดจากการกระทำของผู้ใช้ครั้งเดียวที่ทำให้ต้องแก้ข้อมูลในหลายที่ ด้วย event sourcing คุณออกแบบ event ให้เป็นคำอธิบายที่ครบในตัวเองของการกระทำของผู้ใช้ได้ การกระทำของผู้ใช้จึงต้องการแค่ write เดียวในที่เดียว — คือการต่อท้าย event ลง log — ซึ่งทำให้ atomic ได้ง่าย

ถ้า event log และ application state ถูก shard ในลักษณะเดียวกัน (เช่น การประมวลผล event ของลูกค้าใน shard 3 ต้อง update เฉพาะ shard 3 ของ application state) log consumer แบบ single-thread ธรรมดา ก็ไม่ต้องมี concurrency control สำหรับ write โดยโครงสร้างแล้ว มันประมวลผลทีละ event เท่านั้น (ดู "Actual Serial Execution" ) log กำจัดความไม่แน่นอน (nondeterminism) ของ concurrency โดยกำหนดลำดับแบบ serial ของ event ใน shard [ 27 ] ถ้า event หนึ่งแตะ state shard หลายตัว ก็ต้องทำงานเพิ่มอีกเล็กน้อย ซึ่งเราจะพูดถึงใน บทที่ 13

หลายระบบที่ไม่ได้ใช้ event-sourced model ก็ยังพึ่งพา immutability เพื่อทำ concurrency control ฐานข้อมูลหลายตัวใช้ immutable data structure หรือ multiversion data ภายใน เพื่อรองรับ point-in-time snapshot (ดู "Indexes and snapshot isolation" ) ระบบ version control อย่าง Git, Mercurial และ Fossil ก็พึ่งพา immutable data เพื่อเก็บประวัติเวอร์ชันของไฟล์

Limitations of immutability (ข้อจำกัดของ immutability)

การเก็บประวัติ immutable ของทุกการเปลี่ยนแปลงไว้ตลอดไปทำได้จริงแค่ไหน? คำตอบขึ้นอยู่กับปริมาณการเปลี่ยนแปลงในชุดข้อมูล บาง workload ส่วนใหญ่เพิ่มข้อมูลและไม่ค่อย update หรือ delete ซึ่งทำให้ immutable ได้ง่าย ส่วน workload อื่นมีอัตราการ update และ delete สูง ในชุดข้อมูลที่ค่อนข้างเล็ก ในกรณีเหล่านี้ ประวัติ immutable อาจโตจนใหญ่เกินไป อาจเกิดปัญหา fragmentation และประสิทธิภาพของ compaction และ garbage collection จะสำคัญมากต่อความ robust ในการ operate [ 45 , 46 ]

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

ในสถานการณ์เหล่านี้ แค่เพิ่ม event อีกตัวใน log เพื่อระบุว่าข้อมูลก่อนหน้าควรถือว่าถูกลบแล้วนั้นไม่เพียงพอ — คุณต้องการเขียนประวัติใหม่และทำเหมือนว่าข้อมูลนั้นไม่เคยถูกเขียนตั้งแต่แรก ตัวอย่างเช่น Datomic เรียกฟีเจอร์นี้ว่า excision [ 47 ] ส่วนระบบ version control อย่าง Fossil ก็มีแนวคิดคล้ายกันที่เรียกว่า shunning [ 48 ]

การลบข้อมูลให้หมดจริงๆ ยากอย่างน่าประหลาดใจ [ 49 ] เพราะสำเนาอาจอยู่ได้ในหลายที่ Storage engine, filesystem และ SSD มักเขียนลงตำแหน่งใหม่แทนที่จะเขียนทับข้อมูลที่เดิม [ 41 ] และ backup ก็มักถูกทำให้ immutable โดยเจตนา เพื่อป้องกันการลบหรือความเสียหายโดยไม่ตั้งใจ

วิธีหนึ่งที่ทำให้ลบ immutable data ได้คือ crypto-shredding [ 50 ] ข้อมูลที่คุณอาจต้องการลบในอนาคตจะถูกเก็บแบบเข้ารหัสไว้ และเมื่ออยากกำจัดมัน ก็แค่ทิ้ง (forget) encryption key ข้อมูลที่เข้ารหัสยังอยู่ตรงนั้น แต่ไม่มีใครใช้งานมันได้

ในแง่หนึ่ง นี่เป็นแค่การย้ายปัญหาไปที่อื่น ข้อมูลจริงยังคง immutable แต่ key storage ของคุณเป็น mutable ยิ่งกว่านั้น คุณต้องตัดสินใจล่วงหน้าว่าข้อมูลชุดไหนจะเข้ารหัสด้วย key เดียวกัน และเมื่อไหร่จะใช้ key ต่างกัน — ซึ่งเป็นการตัดสินใจที่สำคัญ เพราะภายหลังคุณ crypto-shred ได้ ได้เพียงทั้งหมดหรือไม่เลยของข้อมูลที่เข้ารหัสด้วย key หนึ่งๆ ไม่สามารถทำเป็นบางส่วนได้ การเก็บ key แยกสำหรับข้อมูลทุกรายการจะยุ่งยากเกินไป เพราะ key storage จะใหญ่เท่ากับ primary data storage Scheme ที่ซับซ้อนกว่านี้ เช่น puncturable encryption [ 51 ] ทำให้เพิกถอนความสามารถในการถอดรหัสของ key ได้แบบเลือกเฉพาะส่วน แต่ยังไม่แพร่หลาย

โดยรวมแล้ว การลบเป็นเรื่องของ "ทำให้การดึงข้อมูลยากขึ้น" มากกว่า "ทำให้เป็นไปไม่ได้ที่จะดึงข้อมูล" แต่บางครั้งเราก็ต้องพยายามอยู่ดี ดังที่เราจะเห็นในหัวข้อ "Legislation and Self-Regulation"

Processing Streams (การประมวลผลสตรีม)

ถึงตอนนี้ในบทนี้ เราได้พูดถึงว่า stream มาจากไหน (event กิจกรรมของผู้ใช้, sensor และ write ลงฐานข้อมูล) และ stream ถูกส่งต่ออย่างไร (ผ่าน direct messaging, ผ่าน message broker และใน event log)

สิ่งที่เหลือคือการพูดถึงว่าเมื่อมี stream แล้วเราทำอะไรกับมันได้ — นั่นคือ คุณสามารถ process (ประมวลผล) มันได้ โดยกว้างๆ แล้วมีสามตัวเลือก:

1. คุณสามารถนำข้อมูลใน event ไปเขียนลงฐานข้อมูล, cache, search index หรือระบบเก็บข้อมูลที่คล้ายกัน เพื่อให้ client อื่น query ต่อได้ ดังแสดงใน Figure 12-5 นี่เป็นวิธีที่ดีในการทำให้ฐานข้อมูล sync กับการเปลี่ยนแปลงที่เกิดขึ้นในส่วนอื่นของระบบ — โดยเฉพาะอย่างยิ่งถ้า stream consumer เป็น client เพียงตัวเดียวที่เขียนลงฐานข้อมูล การเขียนลงระบบเก็บข้อมูลเป็นรูปแบบที่เทียบเท่าในโลก streaming ของสิ่งที่เราพูดถึงในหัวข้อ "Batch Use Cases"

2. คุณสามารถ push event ไปหาผู้ใช้ด้วยวิธีใดวิธีหนึ่ง — เช่น ส่ง email alert หรือ push notification หรือ stream event ไปยัง real-time dashboard ที่แสดงผลเป็นภาพ ในกรณีนี้ มนุษย์เป็น consumer ปลายทางของ stream

3. คุณสามารถประมวลผล input stream ตั้งแต่หนึ่งตัวขึ้นไป เพื่อสร้าง output stream ตั้งแต่หนึ่งตัวขึ้นไป Stream อาจผ่าน pipeline ที่ประกอบด้วยขั้นตอนประมวลผลหลายขั้น ก่อนจะไปจบที่ output (ตัวเลือก 1 หรือ 2)

ในส่วนที่เหลือของบทนี้ เราจะพูดถึงตัวเลือก 3: การประมวลผล stream เพื่อสร้าง derived stream อื่น โค้ดที่ประมวลผล stream แบบนี้เรียกว่า operator หรือ job ซึ่งเกี่ยวข้องใกล้ชิดกับ Unix process และ MapReduce job ที่เราพูดถึงใน บทที่ 11 และ pattern ของ dataflow ก็คล้ายกัน: stream processor consume input stream แบบ read-only และเขียน output ไปยังอีกที่หนึ่งแบบ append-only

Pattern สำหรับ sharding และ parallelization ใน stream processor ก็คล้ายกันมากกับของ MapReduce และ dataflow engine ที่เราเห็นใน บทที่ 11 เราจึงไม่พูดซ้ำเรื่องเหล่านี้ Basic mapping operation เช่น การแปลงและกรอง record ก็ทำงานเหมือนกัน

ความต่างที่สำคัญอย่างหนึ่งจาก batch job คือ stream ไม่มีวันจบ ความต่างนี้มีผลตามมาหลายอย่าง ตามที่กล่าวไว้ในตอนต้นของบทนี้ การ sort ไม่สมเหตุสมผลกับ unbounded dataset จึงใช้ sort-merge join (ดู "Joins and Grouping" ) ไม่ได้ กลไก fault tolerance ก็ต้องเปลี่ยนด้วย กับ batch job ที่รันมาไม่กี่นาที task ที่ล้มเหลวก็ restart ใหม่ตั้งแต่ต้นได้เลย แต่กับ stream job ที่รันมาหลายปี การ restart ตั้งแต่ต้นหลัง crash อาจไม่ใช่ทางเลือกที่เป็นไปได้

Uses of Stream Processing (การใช้งานการประมวลผลสตรีม)

Stream processing ถูกใช้เพื่อ monitor มาเป็นเวลานาน โดยที่องค์กรต้องการให้มีการแจ้งเตือนเมื่อบางสิ่งเกิดขึ้น ตัวอย่างเช่น:

• ระบบตรวจจับการฉ้อโกงต้องระบุว่ารูปแบบการใช้งานบัตรเครดิต เปลี่ยนแปลงอย่างไม่คาดคิดหรือไม่ และบล็อกบัตรถ้ามีแนวโน้มว่าถูกขโมย

• ระบบเทรดต้องตรวจสอบการเปลี่ยนแปลงราคาในตลาดการเงิน และทำการเทรดตามกฎที่กำหนด

• ระบบการผลิตต้องตรวจสอบสถานะของเครื่องจักรในโรงงาน และระบุปัญหาได้อย่างรวดเร็วถ้าเกิดการทำงานผิดปกติ

• ระบบทหารและข่าวกรองต้องติดตามกิจกรรมของผู้ที่อาจเป็นผู้รุกราน และส่งสัญญาณเตือนเมื่อมีสัญญาณของการโจมตี

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

Complex event processing (การประมวลผลเหตุการณ์ที่ซับซ้อน)

Complex event processing (CEP) เป็นแนวทางที่พัฒนาขึ้นในทศวรรษ 1990 สำหรับวิเคราะห์ event stream โดยเน้นแอปพลิเคชันที่ต้องค้นหา event pattern บางอย่าง [ 52 ] เหมือนกับที่ regular expression ช่วยให้ค้นหา pattern ของตัวอักษรใน string ได้ CEP ก็ให้คุณกำหนดกฎเพื่อค้นหา pattern ของ event ใน stream

ระบบ CEP มักใช้ query language แบบ declarative ระดับสูง เช่น SQL หรือ GUI เพื่ออธิบาย pattern ของ event ที่ต้องตรวจจับ Query เหล่านี้จะถูกส่งไปยัง processing engine ที่ consume input stream และเก็บ state machine ไว้ภายใน เพื่อทำการ match ตามที่ต้องการ เมื่อพบ match engine จะปล่อย (emit) complex event (เหตุการณ์ซับซ้อน — ที่มาของชื่อนั่นเอง) พร้อมรายละเอียดของ event pattern ที่ตรวจพบ [ 53 ]

ในระบบเหล่านี้ ความสัมพันธ์ระหว่าง query กับข้อมูล กลับด้านเมื่อเทียบกับฐานข้อมูลทั่วไป โดยปกติ ฐานข้อมูลจะเก็บข้อมูลไว้ถาวรและมอง query เป็นของชั่วคราว เมื่อ query เข้ามา ฐานข้อมูลจะค้นหาข้อมูลที่ตรงกับ query แล้วลืม query นั้นเมื่อทำเสร็จ CEP engine สลับบทบาทเหล่านี้ Query ถูกเก็บไว้ระยะยาว เมื่อแต่ละ event มาถึง engine จะตรวจว่าตอนนี้เจอ event pattern ที่ตรงกับ standing query ใดๆ ของมันแล้วหรือไม่ [ 54 ]

ตัวอย่างการ implement CEP ได้แก่ Esper, Apama และ TIBCO StreamBase Distributed stream processor อย่าง Flink และ Spark Streaming ก็รองรับ SQL สำหรับ declarative query บน stream เช่นกัน

Stream analytics (การวิเคราะห์สตรีม)

Stream processing ยังใช้สำหรับ analytics (การวิเคราะห์) บน streams เส้นแบ่งระหว่าง CEP กับ stream analytics นั้นเลือนราง แต่โดยทั่วไป stream analytics จะไม่ค่อยเน้นการตรวจจับลำดับ event เฉพาะเจาะจง แต่มุ่งไปที่ aggregation และ statistical metrics บน event จำนวนมาก ตัวอย่างแอปพลิเคชันได้แก่:

• การวัดอัตราของ event ประเภทหนึ่ง (เกิดขึ้นบ่อยแค่ไหนต่อช่วงเวลา)

• การคำนวณ rolling average (ค่าเฉลี่ยเคลื่อนที่) ของค่าหนึ่งๆ ภายในช่วงเวลาหนึ่ง

• การเปรียบเทียบสถิติปัจจุบันกับช่วงเวลาก่อนหน้า (เช่น เพื่อตรวจจับแนวโน้ม หรือแจ้งเตือนเมื่อ metrics สูงหรือต่ำผิดปกติเมื่อเทียบกับช่วงเวลาเดียวกันเมื่อสัปดาห์ที่แล้ว)

สถิติเหล่านี้มักคำนวณในช่วงเวลาคงที่ — เช่น คุณอาจอยากรู้จำนวน query ต่อวินาทีโดยเฉลี่ย ของ service ในช่วงห้านาทีที่ผ่านมา และ response time ที่ percentile ที่ 99 ในช่วงเวลานั้น การหาค่าเฉลี่ยในช่วงไม่กี่นาทีช่วยลดความผันผวนที่ไม่สำคัญจากวินาทีต่อวินาที ขณะเดียวกันก็ยังเห็นภาพการเปลี่ยนแปลงของ traffic pattern ได้ทันเวลา ช่วงเวลาที่ใช้ทำ aggregation เรียกว่า window (หน้าต่าง) เราจะดู windowing อย่างละเอียดขึ้นในหัวข้อ "Reasoning About Time"

ระบบ stream analytics บางครั้งใช้ probabilistic algorithm เช่น Bloom filter (ที่เราเจอในหัวข้อ "Bloom filters" ) สำหรับ set membership, HyperLogLog [ 55 ] สำหรับ cardinality estimation, และอัลกอริทึมประมาณค่า percentile ต่างๆ (ดู "Computing Percentiles" ) Probabilistic algorithm ให้ผลลัพธ์โดยประมาณ แต่มีข้อดีคือใช้ memory ใน stream processor น้อยกว่าอัลกอริทึมที่ให้ผลแม่นยำมาก การใช้อัลกอริทึมประมาณค่าแบบนี้ทำให้บางครั้งคนเชื่อว่าระบบ stream processing lossy และไม่แม่นยำเสมอ แต่นั่นเข้าใจผิด stream processing ไม่ได้เป็นการประมาณโดยธรรมชาติแต่อย่างใด และการใช้ probabilistic algorithm เป็นเพียงการ optimize เท่านั้น [ 56 ]

distributed stream processing framework แบบโอเพนซอร์สหลายตัว เช่น Apache Storm, Spark Streaming, Flink, Samza, Apache Beam, และ Kafka Streams ถูกออกแบบโดยคำนึงถึง analytics เป็นหลัก [ 57 ] ส่วนบริการแบบ hosted ได้แก่ Google Cloud Dataflow และ Azure Stream Analytics

Maintaining materialized views (การดูแล materialized view)

เราเห็นแล้วว่า stream ของการเปลี่ยนแปลงในฐานข้อมูล ใช้ทำให้ derived data system เช่น cache, search index และ data warehouse ทันสมัยตรงกับ source database ได้ ทั้งหมดนี้คือตัวอย่างของการดูแล materialized view: คือการ derive view อีกแบบหนึ่งของชุดข้อมูล เพื่อให้ query ได้อย่างมีประสิทธิภาพ และ update view นั้นทุกครั้งที่ข้อมูลพื้นฐานเปลี่ยนแปลง [ 37 ]

ในทำนองเดียวกัน ใน event sourcing application state ถูกดูแลโดยการ apply log ของ event ซึ่ง application state ก็เป็น materialized view ชนิดหนึ่งเช่นกัน ต่างจากสถานการณ์ stream analytics การดูเฉพาะ event ในช่วงเวลาหนึ่งมักไม่เพียงพอ การสร้าง materialized view อาจต้องใช้ event ทั้งหมด ในช่วงเวลาใดก็ได้ ยกเว้น event ที่ล้าสมัยซึ่งอาจถูกทิ้งโดย log compaction กล่าวคือ ต้องมี window ที่ย้อนกลับไปจนถึงจุดเริ่มต้นของเวลา

ในหลักการ stream processor ตัวใดก็ใช้ ดูแล materialized view ได้ แม้ว่าความจำเป็นที่ต้องเก็บ event ไว้ตลอดไป จะขัดกับสมมติฐานของ framework ที่เน้น analytics บางตัว ซึ่งส่วนใหญ่ทำงานบน window ที่มีระยะเวลาจำกัด Kafka Streams และ ksqlDB ของ Confluent รองรับการใช้งานแบบนี้ โดยต่อยอดจากการรองรับ log compaction ของ Kafka [ 58 ]

Incremental View Maintenance (การดูแล view แบบ incremental)

ฐานข้อมูล อาจดูเหมาะกับการดูแล materialized view เพราะอย่างไรเสียมันก็ถูกออกแบบมาให้เก็บสำเนาทั้งหมดของชุดข้อมูล หลายตัวยังรองรับ materialized view ด้วย เราเห็นในหัวข้อ "Materialized Views and Data Cubes" ว่า analytical query ที่พบทั่วไปใน data warehouse สามารถ materialize เป็น OLAP cube ได้

น่าเสียดายที่ฐานข้อมูลมัก refresh ตาราง materialized view ด้วย periodic batch job หรือ on-demand request เช่น REFRESH MATERIALIZED VIEW ของ PostgreSQL ไม่ได้ refresh ทุกครั้งที่ข้อมูลต้นทางถูก update แนวทางนี้มีข้อเสียสำคัญสองข้อที่ทำให้ไม่เหมาะ กับการดูแล view ด้วย stream processing:

ประสิทธิภาพต่ำ (Poor efficiency)

ข้อมูลทั้งหมดถูกประมวลผลใหม่ทุกครั้งที่ view ถูก update ทั้งที่ข้อมูลส่วนใหญ่น่าจะไม่เปลี่ยนแปลง

ความสดใหม่ของข้อมูล (Data freshness)

การเปลี่ยนแปลงในข้อมูลต้นทางจะไม่ปรากฏใน materialized view จนกว่า query ของมันจะถูกรันอีกครั้ง ในรอบ update ตามกำหนดครั้งถัดไป

เราเขียน database trigger ได้ ที่ update materialized view ได้อย่างมีประสิทธิภาพ เมื่อข้อมูลแบ่ง partition ได้ง่าย และการคำนวณเป็น incremental โดยธรรมชาติ เช่น ถ้า materialized view เก็บรายได้จากการขายรวมต่อวัน ก็ update แถวของวันที่ที่เกี่ยวข้องทุกครั้งที่มีการขายใหม่เกิดขึ้นได้ โซลูชันที่ทำเฉพาะกิจใช้ได้ในบางกรณี แต่ SQL query จำนวนมากแปลงเป็น incremental computation ไม่ได้ง่ายๆ หรือไม่ได้อย่างมีประสิทธิภาพ

Incremental view maintenance (IVM) เป็นโซลูชันที่ครอบคลุมกว่าสำหรับปัญหาที่เพิ่งอธิบายไป เทคนิค IVM แปลง query ที่เขียนด้วย SQL หรือภาษาอื่น ให้เป็น operator ที่คำนวณแบบ incremental ได้ แทนที่จะประมวลผลชุดข้อมูลทั้งหมด อัลกอริทึม IVM จะคำนวณใหม่และ update เฉพาะข้อมูลที่เปลี่ยนแปลง [ 38 , 59 , 60 ] การคำนวณ view จึงมีประสิทธิภาพขึ้นมาก ซึ่งหมายความว่ารัน update ได้บ่อยขึ้นมาก และทำให้ข้อมูลสดใหม่ขึ้นอย่างมาก

ฐานข้อมูลอย่าง Materialize [ 61 ], RisingWave, ClickHouse และ Feldera ล้วนใช้เทคนิค IVM เพื่อให้ materialized view แบบ incremental ที่มีประสิทธิภาพ ฐานข้อมูลเหล่านี้ ingest stream ของ event เพื่อให้ใช้ materialized view แบบ real-time ได้ Event ล่าสุดถูก buffer ไว้ใน memory และใช้ update materialized view บน disk เป็นระยะ การอ่านจะรวม event ล่าสุดกับข้อมูลที่ materialize ไว้d เพื่อให้เป็น view แบบ real-time เดียว เนื่องจากการอ่านมักเขียนเป็น SQL และ materialized view มักถูกเก็บในรูปแบบ OLAP-style ระบบเหล่านี้จึงรองรับ query แบบ data warehouse ขนาดใหญ่ด้วย เช่นที่กล่าวถึงใน บทที่ 11

Search on streams (การค้นหาบน stream)

นอกจาก CEP ที่ให้ค้นหา pattern ที่ประกอบด้วยหลาย event บางครั้งก็ต้องค้นหา event แต่ละตัว ตามเกณฑ์ที่ซับซ้อนด้วย เช่น full-text search query ตัวอย่างเช่น บริการ media monitoring จะ subscribe feed ของบทความข่าว และรายการออกอากาศจากสื่อต่างๆ แล้วค้นหาข่าวที่กล่าวถึงบริษัท ผลิตภัณฑ์ หรือหัวข้อที่สนใจ ทำได้โดยกำหนด search query ไว้ล่วงหน้า แล้วจับคู่ stream ของข่าวกับ query นั้นอย่างต่อเนื่อง บางเว็บไซต์ก็มีฟีเจอร์คล้ายกัน — เช่น ผู้ใช้เว็บไซต์อสังหาริมทรัพย์ขอให้แจ้งเตือนได้ เมื่อมีทรัพย์สินใหม่ที่ตรงกับเกณฑ์การค้นหาปรากฏในตลาด ฟีเจอร์ percolator ของ Elasticsearch [ 62 ] เป็นหนึ่งในตัวเลือกสำหรับ implement การค้นหาบน stream แบบนี้

Search engine ทั่วไป index เอกสารก่อน แล้วค่อยรัน query ผ่าน index ในทางตรงกันข้าม การค้นหาบน stream จะกลับด้านการประมวลผล Query ถูกเก็บไว้ และเอกสารถูกนำมาประเมินกับ query เหล่านั้น เหมือนใน CEP ในกรณีที่ง่ายที่สุด คุณทดสอบทุกเอกสารกับทุก query ได้ แม้อาจช้าถ้ามี query จำนวนมาก เพื่อ optimize กระบวนการนี้ เราสามารถ index ตัว query เหมือนกับ index เอกสาร เพื่อจำกัดชุดของ query ที่อาจ match [ 63 ]

Event-driven architectures and RPC (สถาปัตยกรรมแบบ event-driven และ RPC)

ในหัวข้อ "Event-Driven Architectures" เราได้พูดถึงระบบ message-passing ในฐานะทางเลือกแทน RPC กลไกการสื่อสารระหว่าง service แบบนี้ใช้อยู่ เช่น ใน actor model

แม้ระบบเหล่านี้จะอิงกับ message และ event เช่นกัน แต่โดยปกติเราไม่ได้มองว่ามันเป็น stream processor ด้วยเหตุผลบางประการ:

• Actor framework เป็นกลไกสำหรับจัดการ concurrency เป็นหลัก และการ execute แบบกระจายของ module ที่สื่อสารกัน ส่วน stream processing เป็นเทคนิคจัดการข้อมูลเป็นหลัก

• การสื่อสารระหว่าง actor มักเป็น ephemeral และ one-to-one ส่วน event log เป็น durable และมี subscriber ได้หลายราย

• Actor สื่อสารกันแบบใดก็ได้ (รวมถึง request/response แบบ cyclic) แต่ stream processor มักถูกตั้งค่าเป็น acyclic pipeline ที่ทุก stream เป็น output ของ job ใด job หนึ่ง และ derive มาจากชุด input stream ที่กำหนดไว้ชัดเจน

อย่างไรก็ตาม ระบบแบบ RPC กับ stream processing ก็มีส่วนที่คาบเกี่ยวกัน เช่น Apache Storm มีฟีเจอร์ที่เรียกว่า distributed RPC ซึ่งอนุญาตให้กระจาย user query ไปยังชุดของ node ที่ประมวลผล event stream อยู่ด้วย จากนั้น query เหล่านี้จะถูก interleave กับ event จาก input stream และผลลัพธ์จะถูกรวมแล้วส่งกลับไปยังผู้ใช้ได้

เรายังประมวลผล stream ด้วย actor framework ได้เช่นกัน แต่ framework แบบนี้จำนวนมากไม่รับประกันการส่ง message ในกรณีที่เกิด crash การประมวลผลจึงไม่ fault-tolerant เว้นแต่คุณจะ implement retry logic เพิ่มเติมเอง

Reasoning About Time (การคิดเรื่องเวลาอย่างรอบคอบ)

Stream processor มักต้องจัดการกับเรื่องเวลา โดยเฉพาะเมื่อรันงาน analytics ซึ่งมักใช้ time window เช่น "ค่าเฉลี่ยในช่วงห้านาทีที่ผ่านมา" ความหมายของ "ห้านาทีที่ผ่านมา" อาจดูชัดเจนไม่กำกวม แต่น่าเสียดายที่แนวคิดนี้ยุ่งยากอย่างน่าประหลาดใจ

ใน batch process task ประมวลผลจะไล่ crunch ผ่าน event ในอดีตชุดใหญ่อย่างรวดเร็ว ถ้าต้องแบ่งข้อมูลตามเวลา batch process ต้องดู timestamp ที่ฝังอยู่ในแต่ละ event การดู system clock ของเครื่องที่รัน process ไม่มีประโยชน์ เพราะเวลาที่รันไม่เกี่ยวกับเวลาที่ event เกิดขึ้นจริง

Batch process อาจอ่าน event ในอดีตหนึ่งปีเต็มได้ภายในไม่กี่นาที ในกรณีส่วนใหญ่ timeline ที่เราสนใจคือประวัติหนึ่งปีนั้น ไม่ใช่ไม่กี่นาทีของการประมวลผล ยิ่งกว่านั้น การใช้ timestamp ใน event ทำให้การประมวลผลเป็น deterministic: การรัน process เดียวกันอีกครั้งบน input เดียวกันให้ผลลัพธ์เดียวกัน

ในทางกลับกัน framework stream processing หลายตัว ใช้ system clock ของเครื่องที่ประมวลผล (เรียกว่า processing time — เวลาประมวลผล) เพื่อกำหนด windowing [ 64 ] แนวทางนี้มีข้อดีคือเรียบง่าย และสมเหตุสมผลถ้าความล่าช้าระหว่างการเกิด event กับการประมวลผล event สั้นมากจนแทบไม่เป็นปัญหา แต่จะใช้ไม่ได้เมื่อ processing lag มีนัยสำคัญ (คือเมื่อการประมวลผลเกิดช้ากว่าเวลาที่ event เกิดขึ้นอย่างเห็นได้ชัด)

Event time versus processing time (เวลาเหตุการณ์เทียบกับเวลาประมวลผล)

การประมวลผลอาจล่าช้าได้จากหลายสาเหตุ เช่น queueing, network fault, ปัญหาด้านประสิทธิภาพ ที่ทำให้เกิด contention ใน message broker หรือ processor การ restart ของ stream consumer หรือการประมวลผล event ในอดีตซ้ำระหว่างกู้คืนจาก fault หรือหลังจากแก้บั๊กในโค้ด

ความล่าช้าของ message ยังทำให้ลำดับของ message คาดเดาไม่ได้ด้วย ตัวอย่างเช่น สมมติว่าผู้ใช้ส่ง web request ครั้งแรก (ซึ่ง web server A จัดการ) แล้วส่ง request ที่สอง (ซึ่ง server B จัดการ) A และ B ปล่อย event ที่อธิบาย request ที่แต่ละตัวจัดการ แต่ event ของ B ถึง message broker ก่อน event ของ A ตอนนี้ stream processor จะเห็น event B ก่อน แล้วค่อยเห็น event A ทั้งที่จริงเกิดขึ้นในลำดับตรงข้าม

ถ้าอยากได้ analogy ลองนึกถึงภาพยนตร์ Star Wars ภาค IV ออกฉายในปี 1977, ภาค V ในปี 1980 และภาค VI ในปี 1983 ตามด้วยภาค I, II และ III ในปี 1999, 2002 และ 2005 ตามลำดับ และภาค VII, VIII และ IX ในปี 2015, 2017 และ 2019 [ 65 ] ถ้าคุณดูภาพยนตร์ตามลำดับที่ออกฉาย ลำดับที่คุณ process มันจะไม่สอดคล้องกับลำดับของเนื้อเรื่อง (หมายเลขภาคเหมือน event timestamp ส่วนวันที่คุณดูภาพยนตร์คือ processing time) มนุษย์เรารับมือกับความไม่ต่อเนื่องแบบนี้ได้ แต่อัลกอริทึม stream processing ต้องถูกเขียนขึ้นมาโดยเฉพาะ เพื่อรองรับปัญหาด้านเวลาและลำดับเหล่านี้

การสับสนระหว่าง event time กับ processing time ทำให้ได้ข้อมูลที่ผิดเพี้ยน ตัวอย่างเช่น สมมติว่าคุณมี stream processor ที่วัดอัตราของ request (นับจำนวน request ต่อวินาที) ถ้าคุณ redeploy stream processor มันอาจถูกปิดไปหนึ่งนาที แล้วประมวลผล backlog ของ event เมื่อกลับมาทำงาน ถ้าคุณวัดอัตราตาม processing time ก็จะดูเหมือนว่ามี request พุ่งสูง (spike) ผิดปกติอย่างกะทันหัน ระหว่างประมวลผล backlog ทั้งที่จริงๆ แล้วอัตรา request จริงคงที่ (Figure 12-8)

Figure 12-8. Windowing ตาม processing time ทำให้เกิดค่าผิดเพี้ยน (artifact) เนื่องจากอัตราการประมวลผลที่ไม่คงที่

Handling straggler events (การจัดการเหตุการณ์ที่มาช้า)

ปัญหาที่ยุ่งยากเมื่อกำหนด window ตาม event time คือคุณไม่มีทางแน่ใจว่าได้รับ event ทั้งหมดของ window ใด window หนึ่งแล้ว หรือยังมีบางส่วนที่ยังมาไม่ถึง

ตัวอย่างเช่น สมมติว่าคุณจัดกลุ่ม event เป็น window ละ 1 นาที เพื่อนับจำนวน request ต่อนาที คุณนับ event ที่มี timestamp อยู่ในนาทีที่ 37 ของชั่วโมงไปแล้วจำนวนหนึ่ง และเวลาก็เดินต่อไป ตอนนี้ event ที่เข้ามาส่วนใหญ่อยู่ในนาทีที่ 38 และ 39 ของชั่วโมง คุณจะประกาศว่า window ของนาทีที่ 37 เสร็จแล้วเมื่อไหร่ และส่งค่า counter ออกไป?

คุณอาจใช้ timeout แล้วประกาศว่า window พร้อมแล้ว หลังจากไม่เห็น event ใหม่ของ window นั้นมาสักระยะ แต่ event บางตัวอาจยังถูก buffer อยู่บนเครื่องอื่นที่ไหนสักแห่ง และล่าช้าเพราะเครือข่ายสะดุด คุณต้องรับมือกับ straggler event ที่มาถึงหลังจากประกาศว่า window เสร็จสมบูรณ์ไปแล้วได้ โดยกว้างๆ แล้วมีสองทางเลือก [ 1 ]:

• ไม่สนใจ straggler event เพราะน่าจะเป็นสัดส่วนเล็กน้อยของ event ในสถานการณ์ปกติ คุณติดตามจำนวน event ที่ถูก drop เป็น metric ได้ และตั้ง alert ถ้าเริ่ม drop ข้อมูลเป็นปริมาณมาก

• เผยแพร่ correction (การแก้ไข) ซึ่งเป็นค่าที่ update แล้วสำหรับ window โดยรวม straggler เข้าไปด้วย คุณอาจต้องเพิกถอน output ก่อนหน้านั้นด้วย

ในบางกรณี เราใช้ message พิเศษเพื่อระบุได้ว่า "จากนี้ไป จะไม่มี message ที่มี timestamp ก่อน t อีก" ซึ่ง consumer ใช้เพื่อ trigger window ได้ [ 66 ] แต่ถ้ามี producer หลายตัวบนเครื่องต่างๆ สร้าง event โดยแต่ละตัวมี threshold ของ timestamp ขั้นต่ำเป็นของตัวเอง consumer ต้องติดตามแต่ละ producer แยกกัน การเพิ่มและลบ producer ในกรณีนี้ยุ่งยากกว่า

Whose clock are you using, anyway? (แล้วคุณใช้นาฬิกาของใครกันแน่?)

การกำหนด timestamp ให้ event ยากยิ่งกว่านั้นอีก เมื่อ event ถูก buffer ได้หลายจุดในระบบ ตัวอย่างเช่น แอปมือถือที่รายงาน event ของ usage metric ไปยังเซิร์ฟเวอร์ แอปอาจถูกใช้ขณะที่อุปกรณ์ออฟไลน์อยู่ ซึ่งในกรณีนี้จะ buffer event ไว้ในเครื่อง แล้วส่งไปยังเซิร์ฟเวอร์เมื่อมีอินเทอร์เน็ตอีกครั้ง (ซึ่งอาจเป็นหลายชั่วโมงหรือแม้แต่หลายวันต่อมา) สำหรับ consumer ใดๆ ของ stream นี้ event เหล่านี้จะดูเหมือน straggler ที่ล่าช้ามาก

ในบริบทนี้ timestamp ของ event ควรเป็นเวลาที่ผู้ใช้โต้ตอบจริง ตามนาฬิกาในอุปกรณ์มือถือ แต่นาฬิกาบนอุปกรณ์ที่ผู้ใช้ควบคุมมักไว้ใจไม่ได้ เพราะอาจถูกตั้งเวลาผิดโดยไม่ตั้งใจหรือโดยเจตนา (ดู "Clock Synchronization and Accuracy" ) เวลาที่เซิร์ฟเวอร์รับ event (ตามนาฬิกาของเซิร์ฟเวอร์) มีแนวโน้มแม่นยำกว่า เพราะเซิร์ฟเวอร์อยู่ในการควบคุมของคุณ แต่สื่อความหมายได้น้อยกว่าในแง่การบอกว่าผู้ใช้โต้ตอบเมื่อไหร่

เพื่อชดเชยนาฬิกาอุปกรณ์ที่ไม่ถูกต้อง วิธีหนึ่งคือบันทึก timestamp สามค่า [ 67 ]:

• เวลาที่ event เกิดขึ้น ตามนาฬิกาอุปกรณ์

• เวลาที่ event ถูกส่งไปยังเซิร์ฟเวอร์ ตามนาฬิกาอุปกรณ์

• เวลาที่ event ถูกรับโดยเซิร์ฟเวอร์ ตามนาฬิกาเซิร์ฟเวอร์

เมื่อลบ timestamp ที่สองออกจากที่สาม ก็ประมาณค่า offset ระหว่างนาฬิกาอุปกรณ์กับนาฬิกาเซิร์ฟเวอร์ได้ (โดยสมมติว่า network delay น้อยมากเมื่อเทียบกับความแม่นยำของ timestamp ที่ต้องการ) จากนั้นนำ offset นั้นไปปรับ event timestamp เพื่อประมาณเวลาจริงที่ event เกิดขึ้น (โดยสมมติว่า offset ของนาฬิกาอุปกรณ์ไม่เปลี่ยนระหว่างเวลาที่ event เกิดขึ้น กับเวลาที่ถูกส่งไปยังเซิร์ฟเวอร์)

ปัญหานี้ไม่ได้เกิดเฉพาะกับ stream processing batch processing ก็เจอปัญหาเรื่องการคิดเกี่ยวกับเวลาแบบเดียวกันเป๊ะ เพียงแต่เห็นได้ชัดกว่าในบริบท streaming ที่เราตระหนักถึงการผ่านไปของเวลามากกว่า

Types of windows (ประเภทของ window)

เมื่อรู้แล้วว่าควรกำหนด timestamp ของ event อย่างไร ขั้นตอนถัดไปคือตัดสินใจว่าจะกำหนด window ตามช่วงเวลาอย่างไร จากนั้นเราใช้ window ทำ aggregation ได้ — เช่น นับ event หรือคำนวณค่าเฉลี่ยของค่าภายใน window window หลายประเภทเป็นที่ใช้กันทั่วไป [ 64 , 68 ]:

Tumbling windows (window ที่ต่อกันโดยไม่ซ้อนทับ)

Tumbling window มีความยาวคงที่ และทุก event อยู่ใน window เดียวเท่านั้น ตัวอย่างเช่น ถ้ามี tumbling window หนึ่งนาที event ทั้งหมดที่มี timestamp ตั้งแต่ 10:03:00 ถึง 10:03:59 จะถูกจัดเป็น window เดียว event ตั้งแต่ 10:04:00 ถึง 10:04:59 อยู่ใน window ถัดไป เป็นต้น คุณ implement tumbling window หนึ่งนาทีได้ โดยนำ timestamp ของแต่ละ event มาปัดลงเป็นนาทีที่ใกล้ที่สุด เพื่อกำหนดว่ามันอยู่ใน window ไหน

Hopping windows (window ที่กระโดดเป็นช่วง)

Hopping window ก็มีความยาวคงที่เช่นกัน แต่ window ที่อยู่ติดกันจะซ้อนทับกันเพื่อให้ผลดูเรียบขึ้น (smoothing) ตัวอย่างเช่น window ห้านาทีที่มี hop size หนึ่งนาที จะมี event ตั้งแต่ 10:03:00 ถึง 10:07:59 จากนั้น window ถัดไปครอบคลุม event ตั้งแต่ 10:04:00 ถึง 10:08:59 เป็นต้น คุณ implement ได้โดยคำนวณ tumbling window หนึ่งนาทีก่อน แล้วรวมผลของหลาย window ที่อยู่ติดกัน

Sliding windows (window แบบเลื่อน)

Sliding window ครอบคลุม event ทั้งหมดที่เกิดขึ้น ภายในช่วงเวลาหนึ่งจากกันและกัน ตัวอย่างเช่น sliding window ห้านาทีจะครอบคลุม event ที่ 10:03:39 และ 10:08:12 เพราะทั้งสองห่างกันน้อยกว่าห้านาที (สังเกตว่า tumbling และ hopping window ห้านาที จะไม่จัด event สองตัวนี้ไว้ใน window เดียวกัน เพราะใช้ขอบเขตตายตัว) Sliding window implement ได้โดยเก็บ buffer ของ event ที่เรียงตามเวลา แล้วลบ event เก่า ออกเมื่อหมดอายุจาก window

Session windows (window แบบ session)

ต่างจาก window ประเภทอื่น session window ไม่มีระยะเวลาคงที่ แต่ถูกกำหนดโดยการจัดกลุ่ม event ทั้งหมดของผู้ใช้คนเดียวกัน ที่เกิดขึ้นใกล้กันในเวลา และ window จะจบเมื่อผู้ใช้ไม่ได้ใช้งานมาระยะหนึ่ง (เช่น ไม่มี event เกิดขึ้นเลย 30 นาที) Sessionization เป็นความต้องการที่พบบ่อย ในงาน analytics ของเว็บไซต์

การดำเนินการแบบ window มักต้องเก็บ state ชั่วคราว ในบางกรณี state มีขนาดคงที่ ไม่ว่า window จะใหญ่แค่ไหนหรือมี event กี่ตัว — เช่น การนับจะมี counter ตัวเดียว ไม่ว่า window หรือจำนวน event จะเป็นเท่าไร ในทางกลับกัน sliding window หรือ stream join ซึ่งเราจะพูดถึงในส่วนถัดไป ต้องให้ event ถูก buffer ไว้จนกว่า window จะจบ ดังนั้น window ขนาดใหญ่หรือ stream ที่มี throughput สูง อาจทำให้ stream processor ต้องเก็บ state ชั่วคราวจำนวนมาก คุณจึงต้องดูแลให้เครื่องที่รันงาน stream processing มีความจุเพียงพอที่จะเก็บ state นี้ ไม่ว่าจะใน memory หรือบน disk

Stream Joins (การ join บน stream)

ในหัวข้อ "Joins and Grouping" เราได้พูดถึงว่า batch job join ชุดข้อมูลด้วย key ได้อย่างไร และ join แบบนี้เป็นส่วนสำคัญของ data pipeline เนื่องจาก stream processing ขยาย data pipeline ไปสู่การประมวลผลแบบ incremental ของ unbounded dataset จึงมีความต้องการ join บน stream แบบเดียวกันเป๊ะ

แต่เพราะ event ใหม่เกิดขึ้นบน stream ได้ทุกเมื่อ การ join บน stream จึงท้าทายกว่าใน batch job เพื่อให้เข้าใจสถานการณ์ดีขึ้น ลองแยก join ออกเป็นสามประเภท: stream–stream join, stream–table join และ table–table join ในหัวข้อถัดไป เราจะอธิบายแต่ละประเภทพร้อมตัวอย่าง

Stream–stream join (window join) (การ join stream กับ stream)

สมมติว่าเว็บไซต์ของคุณมีฟีเจอร์ค้นหา และคุณอยากตรวจจับแนวโน้มล่าสุดของ URL ที่ถูกค้นหา ทุกครั้งที่มีคนพิมพ์ search query คุณ log event ที่มี query และผลลัพธ์ที่ส่งกลับ ทุกครั้งที่มีคนคลิกผลการค้นหาหนึ่งรายการ คุณ log อีก event ที่บันทึกการคลิก ในการคำนวณ click-through rate (อัตราการคลิกผ่าน) ของแต่ละ URL ในผลการค้นหา คุณต้องนำ event ของการค้นหาและ event ของการคลิกมารวมกัน ซึ่งเชื่อมโยงกันด้วย session ID เดียวกัน ระบบโฆษณาก็ต้องมีการวิเคราะห์แบบเดียวกัน [ 69 ]

การคลิกอาจไม่เกิดขึ้นเลยถ้าผู้ใช้เลิกค้นหากลางคัน และแม้จะมีการคลิก เวลาระหว่างการค้นหากับการคลิกก็อาจต่างกันมาก ในหลายกรณีอาจเป็นแค่ไม่กี่วินาที แต่ก็อาจนานเป็นวันหรือสัปดาห์ (ถ้าผู้ใช้ค้นหาแล้วลืมแท็บเบราว์เซอร์นั้น ก่อนจะกลับมาที่แท็บแล้วคลิกผลลัพธ์ทีหลัง) เพราะ network delay ที่ไม่คงที่ click event อาจมาถึงก่อน search event ด้วยซ้ำ คุณเลือก window ที่เหมาะสมสำหรับ join ได้ — เช่น คุณอาจเลือก join การคลิกกับการค้นหา ถ้าทั้งสองเกิดห่างกันไม่เกินหนึ่งชั่วโมง

สังเกตว่าการฝังรายละเอียดของการค้นหาไว้ใน click event ไม่เทียบเท่ากับการ join event วิธีนั้นจะบอกได้เฉพาะกรณีที่ผู้ใช้คลิกผลการค้นหา ไม่บอกอะไรเกี่ยวกับการค้นหาที่ผู้ใช้ไม่ได้คลิกผลลัพธ์ใดเลย ในการวัดคุณภาพการค้นหา คุณต้องมี click-through rate ที่แม่นยำ ซึ่งต้องใช้ทั้ง search event และ click event

เพื่อ implement join ประเภทนี้ stream processor ต้องเก็บ state — เช่น event ทั้งหมดที่เกิดขึ้นในชั่วโมงที่ผ่านมา โดย index ตาม session ID ทุกครั้งที่เกิด search event หรือ click event event นั้นจะถูกเพิ่มลงใน index ที่เหมาะสม และ stream processor ก็ตรวจ index อีกอัน ว่ามี event อื่นของ session ID เดียวกันมาถึงแล้วหรือยัง ถ้ามี event ที่ตรงกัน คุณ emit event ที่บอกว่าผลการค้นหาไหนถูกคลิก ถ้า search event หมดอายุโดยไม่เจอ click event ที่ตรงกัน คุณ emit event ที่บอกว่าผลการค้นหาไหนไม่ถูกคลิก

Stream–table join (stream enrichment) (การ join stream กับ table)

ในหัวข้อ "Joins and Grouping" ( Figure 11-2 ) เราเห็นตัวอย่าง batch job ที่ join ข้อมูลสองชุด: user activity event ชุดหนึ่งกับฐานข้อมูล user profile เป็นเรื่องธรรมชาติที่จะมอง user activity event เป็น stream แล้วทำ join แบบเดียวกันอย่างต่อเนื่องใน stream processor Input คือ stream ของ activity event ที่มี user ID และ output คือ stream ของ activity event ที่ user ID ถูกเสริม (augment) ด้วยข้อมูล profile ของผู้ใช้ กระบวนการนี้บางครั้งเรียกว่า enriching (การเพิ่มคุณค่า) activity event ด้วยข้อมูลจากฐานข้อมูล

ในการทำ join นี้ stream processor ต้องรับ activity event ทีละตัว ค้นหา user ID ของ event นั้นในฐานข้อมูล แล้วเพิ่มข้อมูล profile ลงใน activity event การค้นหาในฐานข้อมูลอาจ implement ด้วยการ query ฐานข้อมูลระยะไกล แต่ตามที่กล่าวไว้ในหัวข้อ "Joins and Grouping" query ระยะไกลแบบนี้มีแนวโน้มช้า และเสี่ยงทำให้ฐานข้อมูล overload [ 58 ]

อีกแนวทางคือโหลดสำเนาของฐานข้อมูลเข้าไปใน stream processor เพื่อ query ในเครื่องได้โดยไม่ต้องไป-กลับผ่านเครือข่าย (round trip) เทคนิคนี้เรียกว่า hash join เพราะสำเนาในเครื่องของฐานข้อมูลอาจเป็น hash table ใน memory ถ้าเล็กพอ หรืออาจเป็น index บน local disk

ความต่างจาก batch job คือ batch job ใช้ point-in-time snapshot ของฐานข้อมูลเป็น input ส่วน stream processor ทำงานต่อเนื่องยาวนาน และเนื้อหาของฐานข้อมูลมีแนวโน้มเปลี่ยนไปตามเวลา สำเนาในเครื่องของฐานข้อมูลใน stream processor จึงต้องถูกทำให้ทันสมัยอยู่เสมอ ปัญหานี้แก้ได้ด้วย CDC Stream processor สามารถ subscribe changelog ของฐานข้อมูล user profile ควบคู่ไปกับ stream ของ activity event เมื่อ profile ถูกสร้างหรือแก้ไข stream processor ก็ update สำเนาในเครื่อง ดังนั้นเราจึงได้ join ระหว่างสอง stream: activity event กับ profile update

Stream–table join คล้ายกับ stream–stream join มาก ความต่างที่ใหญ่ที่สุดคือ สำหรับ stream ของ changelog ของ table join จะใช้ window ที่ย้อนไปถึง "จุดเริ่มต้นของเวลา" (window ที่ไม่มีที่สิ้นสุดในเชิงแนวคิด) โดย version ใหม่กว่าของ record จะเขียนทับ version เก่ากว่า ส่วน stream input นั้น join อาจไม่เก็บ window เลย

Table–table join (materialized view maintenance) (การ join table กับ table)

ลองดูตัวอย่าง timeline ของ social network ที่เราพูดถึงในหัวข้อ "Case Study: Social Network Home Timelines" เราบอกว่าเมื่อผู้ใช้ต้องการดู home timeline การไล่ iterate ดูทุกคนที่ผู้ใช้ติดตามนั้นแพงเกินไป ทั้งต้องหาโพสต์ล่าสุดของแต่ละคนแล้วรวมเข้าด้วยกัน

เราจึงต้องการ timeline cache แทน ซึ่งเป็น "กล่องจดหมาย" ประจำผู้ใช้แต่ละคน ที่โพสต์ถูกเขียนลงไปเมื่อมีการส่งโพสต์ เพื่อให้การอ่าน timeline เป็นการ lookup เพียงครั้งเดียว การ materialize และดูแล cache นี้ ต้องมีการประมวลผล event ดังต่อไปนี้:

• เมื่อผู้ใช้ u ส่งโพสต์ใหม่ โพสต์นั้นจะถูกเพิ่มลงใน timeline ของผู้ใช้ทุกคนที่ติดตาม u

• เมื่อผู้ใช้ลบโพสต์ หรือลบบัญชีทั้งหมด โพสต์จะถูกลบออกจาก timeline ของผู้ใช้ทุกคน

• เมื่อผู้ใช้ u 1 เริ่มติดตามผู้ใช้ u 2 โพสต์ล่าสุดของ u 2 จะถูกเพิ่มไปยัง u 1 's timeline

• เมื่อผู้ใช้ u 1 เลิกติดตามผู้ใช้ u 2 โพสต์ของ u 2 จะถูกลบออกจาก u 1 's timeline

ในการ implement การดูแล cache นี้ใน stream processor คุณต้องมี stream ของ event สำหรับโพสต์ (การส่งและการลบ) และสำหรับความสัมพันธ์การติดตาม (การติดตามและการเลิกติดตาม) Stream processor ต้องเก็บฐานข้อมูลที่มีชุดผู้ติดตามของผู้ใช้แต่ละคน เพื่อให้รู้ว่าต้อง update timeline ของใครเมื่อมีโพสต์ใหม่มาถึง

อีกมุมมองหนึ่งของ stream processor ตัวนี้คือ มันดูแล materialized view ของ query ที่ join สองตาราง ( posts และ follows ) — ประมาณนี้:

SELECT follows.follower_id AS timeline_id,
  array_agg(posts.* ORDER BY posts.timestamp DESC)
FROM posts
JOIN follows ON follows.followee_id = posts.sender_id
GROUP BY follows.follower_id

การ join ของ stream สอดคล้องโดยตรงกับการ join ของตารางใน query นี้ Timeline จึงเป็น cache ของผลลัพธ์ของ query นี้โดยปริยาย ที่ถูก update ทุกครั้งที่ตารางพื้นฐานเปลี่ยนแปลง

Note

ถ้ามอง stream เป็นอนุพันธ์ของตาราง ดังใน Figure 12-7 และมอง join เป็นผลคูณ (product) ของสองตาราง u·v จะเห็นสิ่งที่น่าสนใจ: stream ของการเปลี่ยนแปลงของ materialized join เป็นไปตามกฎผลคูณ (product rule) ( u·v )′ =  u ′ v  +  uv ′ การเปลี่ยนแปลงใดๆ ของ posts จะถูก join กับผู้ติดตามปัจจุบัน และการเปลี่ยนแปลงใดๆ ของ follows จะถูก join กับโพสต์ปัจจุบัน [ 37 ]

Time dependence of joins (การพึ่งพาเวลาของ join)

join ทั้งสามประเภทที่อธิบายไว้ที่นี่ (stream–stream, stream–table และ table–table) มีอะไรที่เหมือนกันมาก ทั้งหมดต้องให้ stream processor เก็บ state (search และ click event, user profile หรือรายชื่อ follower) ที่ derive จาก join input ฝั่งหนึ่ง แล้ว query state นั้นเมื่อประมวลผล record จาก input อีกฝั่ง

ลำดับของ event ที่ใช้เก็บ state มีความสำคัญ — เช่น มีผลว่าคุณติดตามก่อนแล้วค่อยเลิกติดตามผู้ใช้ หรือสลับกัน ใน sharded event log เช่น Kafka ลำดับของ event ภายใน shard (partition) เดียวจะถูกรักษาไว้ แต่โดยทั่วไปไม่มีการรับประกันลำดับข้าม stream หรือ shard ที่ต่างกัน

สิ่งนี้ทำให้เกิดคำถามว่า ถ้า event บน stream ที่ต่างกันเกิดขึ้นในเวลาใกล้เคียงกัน จะถูกประมวลผลตามลำดับใด? ยกตัวอย่าง stream–table join ถ้าผู้ใช้ update profile ของตัวเอง activity event ไหนจะถูก join กับ profile เก่า (ที่ประมวลผลก่อน profile update) และ event ไหนจะถูก join กับ profile ใหม่ (ที่ประมวลผลหลัง profile update)? พูดอีกอย่างคือ: ถ้า state เปลี่ยนไปตามเวลา แล้วคุณ join กับ state นั้น คุณจะใช้จุดเวลาไหนสำหรับ join?

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

ถ้าไม่ได้กำหนดลำดับของ event ข้าม stream ไว้ join จะกลายเป็น nondeterministic [ 70 ] ซึ่งหมายความว่าการรัน job เดียวกัน บน input เดียวกันไม่ได้รับประกันว่าจะได้ผลลัพธ์เดียวกันเสมอไป Event บน input stream อาจถูกสลับลำดับ (interleave) ต่างกัน เมื่อรัน job ใหม่อีกครั้ง

ใน data warehouse ปัญหานี้เรียกว่า slowly changing dimension (SCD) และมักแก้ด้วยการใช้ identifier ที่ไม่ซ้ำกัน สำหรับ version เฉพาะของ record ที่ถูก join — เช่น ทุกครั้งที่อัตราภาษีเปลี่ยน ก็จะได้ identifier ใหม่ และใบแจ้งหนี้จะรวม identifier ของอัตราภาษี ณ เวลาที่ขายไว้ [ 71 , 72 ] การเปลี่ยนแปลงนี้ทำให้ join เป็น deterministic แต่มีผลตามมาคือทำ log compaction ไม่ได้ เพราะต้องเก็บ record ทุก version ในตารางไว้ อีกทางหนึ่งคือ denormalize ข้อมูล แล้วใส่อัตราภาษีที่เกี่ยวข้องลงในทุก event การขายโดยตรง

Fault Tolerance (ความทนทานต่อ fault)

ในส่วนสุดท้ายของบทนี้ ลองดูว่า stream processor ทนต่อ fault ได้อย่างไร เราเห็นใน บทที่ 11 ว่า batch processing framework ทนต่อ fault ได้ค่อนข้างง่าย: ถ้า task ล้มเหลว ก็แค่เริ่มใหม่บนเครื่องอื่น และ output ของ task ที่ล้มเหลวจะถูกทิ้ง การ retry แบบโปร่งใสนี้ทำได้เพราะ input file เป็น immutable แต่ละ task เขียน output ลงไฟล์แยกต่างหาก และ output จะ visible ก็ต่อเมื่อ task เสร็จสมบูรณ์เท่านั้น

โดยเฉพาะอย่างยิ่ง แนวทาง batch ในการทนต่อ fault ทำให้แน่ใจว่า output ของ batch job เหมือนกับว่าไม่มีอะไรผิดพลาดเลย แม้บาง task จะล้มเหลวจริงก็ตาม ดูราวกับว่าทุก input record ถูกประมวลผลแบบ exactly once — ไม่มี record ถูกข้าม และไม่มี record ใดถูกประมวลผลสองครั้ง แม้การ restart task จะหมายความว่า record อาจถูกประมวลผลหลายครั้ง แต่ผลที่มองเห็นได้ใน output เหมือนกับว่าถูกประมวลผลเพียงครั้งเดียว หลักการนี้เรียกว่า exactly-once semantics (semantics แบบครั้งเดียวแน่นอน) แม้ว่าคำว่า effectively-once (ครั้งเดียวในทางปฏิบัติ) จะสื่อความหมายได้ตรงกว่า [ 73 ]

ปัญหา fault tolerance แบบเดียวกันเกิดขึ้นใน stream processing แต่จัดการได้ไม่ตรงไปตรงมาเท่า การรอให้ task เสร็จก่อนจึงทำให้ output visible ไม่ใช่ทางเลือก เพราะ stream ไม่มีที่สิ้นสุด จึงไม่มีวันประมวลผลเสร็จ

Microbatching and checkpointing (การทำ batch ย่อยและการ checkpoint)

วิธีแก้วิธีหนึ่งคือแบ่ง stream ออกเป็นบล็อกเล็กๆ แล้วมองแต่ละบล็อกเหมือน batch process ขนาดเล็ก แนวทางนี้เรียกว่า microbatching และใช้ใน Spark Streaming [ 74 ] โดยทั่วไป batch จะมีขนาดประมาณหนึ่งวินาที ซึ่งเป็นผลจากการประนีประนอมด้านประสิทธิภาพ: batch ที่เล็กลงมี overhead ด้าน scheduling และ coordination มากขึ้น ส่วน batch ที่ใหญ่ขึ้นหมายถึงต้องรอนานขึ้น กว่าผลลัพธ์ของ stream processor จะ visible

Microbatching ยังให้ tumbling window โดยปริยาย ที่มีขนาดเท่ากับ batch (window ตาม processing time ไม่ใช่ event timestamp) job ใดที่ต้องการ window ที่ใหญ่กว่านั้น ต้องส่งต่อ state จาก microbatch หนึ่งไปยังอันถัดไปอย่างชัดเจน

แนวทางอีกรูปแบบหนึ่งที่ใช้ใน Apache Flink คือสร้าง rolling checkpoint ของ state เป็นระยะ แล้วเขียนลง durable storage [ 75 , 76 ] ถ้า stream operator crash ก็ restart จาก checkpoint ล่าสุดได้ และทิ้ง output ใดๆ ที่สร้างขึ้นระหว่าง checkpoint ล่าสุดกับตอนที่ crash Checkpoint ถูก trigger โดย barrier ใน message stream คล้ายกับขอบเขตระหว่าง microbatch แต่ไม่ได้บังคับขนาด window เฉพาะเจาะจง

ภายในขอบเขตของ stream processing framework แนวทาง microbatching และ checkpointing ให้ exactly-once semantics แบบเดียวกับ batch processing แต่ทันทีที่ output ออกจาก stream processor (เช่น เมื่อเขียนลงฐานข้อมูล, เผยแพร่ message ไปยัง external message broker, หรือ trigger การส่งอีเมล) framework จะไม่สามารถทิ้ง output ของ microbatch ที่ล้มเหลวได้อีกต่อไป ในกรณีนี้ การ restart task ที่ล้มเหลว ทำให้ side effect ภายนอกเกิดขึ้นสองครั้ง และ microbatching หรือ checkpointing อย่างเดียวไม่เพียงพอ ที่จะป้องกันปัญหานี้

Atomic commit revisited (กลับมาดู atomic commit อีกครั้ง)

เพื่อให้ดูเหมือนว่าการประมวลผลเป็นแบบ exactly-once แม้มี fault เกิดขึ้น เราต้องแน่ใจว่า output และ side effect ทั้งหมด ของการประมวลผล event จะถูก persist ถ้าและต่อเมื่อ การประมวลผลนั้นสำเร็จเท่านั้น ซึ่งรวมถึง message ใดๆ ที่ส่งไปยัง downstream operators หรือระบบ messaging ภายนอก (รวมถึงอีเมลหรือ push notification), database write, การเปลี่ยนแปลง state ของ operator และ acknowledgment ของ input message (รวมถึงการเลื่อน consumer offset ไปข้างหน้าใน log-based message broker)

การกระทำเหล่านี้ต้องเกิดขึ้นแบบ atomic: ไม่ว่าทั้งหมดจะเกิดขึ้น หรือไม่มีเลย ถ้าแนวคิดนี้ฟังดูคุ้นๆ นั่นเป็นเพราะเราเคยพูดถึงมันในหัวข้อ "Exactly-once message processing" ในบริบทของ distributed transaction และ two-phase commit

เราได้พิจารณาปัญหาของการ implement distributed transaction แบบดั้งเดิม เช่น XA ใน บทที่ 8 แต่ในสภาพแวดล้อมที่จำกัดกว่า ก็เป็นไปได้ที่จะ implement atomic commit facility แบบนี้ อย่างมีประสิทธิภาพ แนวทางนี้ใช้ใน Google Cloud Dataflow [ 66 , 75 ], VoltDB [ 77 ] และ Apache Kafka [ 78 , 79 ] ซึ่งต่างจาก XA การ implement เหล่านี้ไม่ได้พยายามให้ transaction ข้ามเทคโนโลยีที่ต่างกัน (heterogeneous) แต่เก็บ transaction ไว้ภายใน โดยจัดการทั้งการเปลี่ยนแปลง state และการส่ง message ภายใน stream processing framework Overhead ของ transaction protocol ถูก amortize ได้โดยการประมวลผล input message หลายรายการ ภายใน transaction เดียว

Idempotence (การทำให้มีผลเหมือนเดิม)

เป้าหมายของเราคือทิ้ง output บางส่วนของ task ที่ล้มเหลว เพื่อให้ retry ได้อย่างปลอดภัย Distributed transaction เป็นวิธีหนึ่งในการบรรลุเป้าหมายนั้น อีกวิธีคือพึ่งพา idempotence (การทำให้มีผลเหมือนเดิม) ดังที่เราเห็นในหัวข้อ "Durable Execution and Workflows" [ 80 ]

การดำเนินการแบบ idempotent คือการดำเนินการที่ทำกี่ครั้งก็ได้ และให้ผลเหมือนกับทำเพียงครั้งเดียว เช่น การลบ key ใน key-value store เป็น idempotent (การลบค่านั้นอีกครั้งไม่มีผลเพิ่มเติม) ส่วนการเพิ่มค่า counter ไม่เป็น idempotent (การเพิ่มอีกครั้งหมายความว่าค่าถูกเพิ่มสองครั้ง)

แม้การดำเนินการจะไม่ idempotent โดยธรรมชาติ ก็มักทำให้เป็น idempotent ได้ ด้วย metadata เพิ่มเติมเล็กน้อย ตัวอย่างเช่น เมื่อ consume message จาก Kafka ทุก message มี offset ที่คงอยู่และเพิ่มขึ้นเรื่อยๆ (monotonically increasing) เมื่อเขียนค่าลงฐานข้อมูลภายนอก คุณรวม offset ของ message ที่ trigger การเขียนครั้งล่าสุดไว้กับค่านั้นได้ คุณจึงบอกได้ว่า update นั้นถูก apply ไปแล้วหรือยัง และเลี่ยงการ update เดียวกันซ้ำ การจัดการ state ใน Trident ของ Storm ก็อาศัยแนวคิดที่คล้ายกัน

การพึ่งพา idempotence มีข้อสมมติหลายอย่างตามมา: การ restart task ที่ล้มเหลวต้อง replay message ชุดเดิมในลำดับเดิม (log-based message broker ทำเช่นนี้) การประมวลผลต้องเป็น deterministic และไม่มี node อื่นอัปเดตค่าเดียวกันพร้อมกันได้ [ 81 , 82 ] เมื่อ failover จาก processing node หนึ่งไปอีก node หนึ่ง อาจต้องใช้ fencing (ดู "Distributed Locks and Leases" ) เพื่อป้องกันการรบกวนจาก node ที่เชื่อกันว่าตายแล้วแต่จริงๆ ยังทำงานอยู่ แต่แม้จะมีข้อควรระวังทั้งหมดนี้ การดำเนินการแบบ idempotent ก็เป็นวิธีที่ได้ผลในการบรรลุ exactly-once semantics ด้วย overhead เพียงเล็กน้อย

Rebuilding state after a failure (การสร้าง state ใหม่หลังเกิด failure)

stream process ใดๆ ที่ต้องใช้ state — เช่น windowed aggregation (อย่าง counter, average และ histogram) และ table กับ index ใดๆ ที่ใช้ใน join — ต้องแน่ใจว่าสามารถกู้คืน state นี้ได้หลังเกิด failure

ตัวเลือกหนึ่งคือเก็บ state ไว้ใน remote datastore แล้ว replicate แม้การต้อง query ฐานข้อมูลระยะไกลสำหรับแต่ละ message อาจช้า อีกทางเลือกคือเก็บ state ไว้ใน stream processor เอง แล้ว replicate เป็นระยะ จากนั้นเมื่อ stream processor กำลังกู้คืนจาก failure task ใหม่ก็อ่าน state ที่ถูก replicate ไว้ แล้วประมวลผลต่อได้โดยไม่สูญเสียข้อมูล

ตัวอย่างเช่น Flink จับ snapshot ของ operator state เป็นระยะ แล้วเขียนลง durable storage เช่น distributed filesystem [ 75 , 76 ] และ Kafka Streams replicate การเปลี่ยนแปลง state โดยส่งไปยัง Kafka topic เฉพาะที่มี log compaction คล้ายกับ CDC [ 83 ] VoltDB replicate state โดยประมวลผล input message แต่ละรายการซ้ำซ้อนบนหลาย node (ดู "Actual Serial Execution" )

ในบางกรณี อาจไม่จำเป็นต้อง replicate state ด้วยซ้ำ เพราะสร้างใหม่จาก input stream ได้ ตัวอย่างเช่น ถ้า state ประกอบด้วย aggregation บน window ที่ค่อนข้างสั้น ก็อาจเร็วพอที่จะ replay input event ที่ตรงกับ window นั้นซ้ำ ถ้า state เป็น local replica ของฐานข้อมูล ที่ดูแลโดย CDC ก็สร้างฐานข้อมูลขึ้นใหม่จาก change stream ที่ทำ log compaction ไว้ได้เช่นกัน

ทั้งหมดนี้ขึ้นอยู่กับลักษณะด้านประสิทธิภาพของ infrastructure พื้นฐาน ในบางระบบ network delay อาจต่ำกว่า latency ของการเข้าถึง disk และ network bandwidth อาจเทียบได้กับ disk bandwidth ไม่มีโซลูชันใดที่เหมาะที่สุดกับทุกสถานการณ์ และข้อดีของ state แบบ local เทียบกับ remote ก็อาจเปลี่ยนไปเมื่อเทคโนโลยี storage และ network พัฒนาขึ้น

Summary (สรุป)

ในบทนี้เราได้พูดถึง event stream, จุดประสงค์ที่มันใช้งาน และวิธีประมวลผล ในบางแง่ stream processing คล้ายกับ batch processing ที่เราพูดถึงใน บทที่ 11 มาก แต่ทำอย่างต่อเนื่องบน stream ที่ไม่มีขอบเขต (ไม่มีวันจบ) แทนที่จะเป็น input ขนาดคงที่ [ 84 ] จากมุมมองนี้ message broker และ event log ทำหน้าที่เทียบเท่ากับ filesystem ในโลก streaming

เราใช้เวลาพอสมควรเปรียบเทียบ message broker สองประเภท:

AMQP/JMS-style message broker

Broker กำหนด message แต่ละตัวให้ consumer และ consumer ยืนยัน (ack) message แต่ละตัวเมื่อประมวลผลสำเร็จ Message จะถูกลบออกจาก broker หลังจากถูก ack แล้ว แนวทางนี้เหมาะเป็น RPC แบบ asynchronous (ดูเพิ่มเติมที่ "Event-Driven Architectures" ) — เช่น ใน task queue ที่ลำดับการประมวลผล message แบบตายตัวไม่สำคัญ และไม่ต้องย้อนกลับไปอ่าน message เก่าอีก หลังจากประมวลผลไปแล้ว

Log-based message broker

Broker กำหนด message ทั้งหมดใน shard ให้ consumer node ตัวเดียวกัน และส่ง message ตามลำดับเดิมเสมอ ทำ parallelism ผ่าน sharding และ consumer ติดตามความคืบหน้าด้วยการ checkpoint offset ของ message ล่าสุดที่ประมวลผลแล้ว Broker เก็บ message ไว้บน disk จึงย้อนกลับไปอ่าน message เก่าซ้ำได้ถ้าจำเป็น

แนวทางแบบ log คล้ายกับ replication log ที่พบในฐานข้อมูล (ดู บทที่ 6 ) และ log-structured storage engine (ดู บทที่ 4 ) และยังเป็นรูปแบบหนึ่งของ consensus ดังที่เราเห็นใน บทที่ 10 เราเห็นว่าแนวทางนี้เหมาะเป็นพิเศษกับระบบ stream processing ที่ consume input stream และสร้าง derived state หรือ derived output stream

ในแง่ของที่มาของ stream เราได้พูดถึงความเป็นไปได้หลายอย่าง event กิจกรรมของผู้ใช้ sensor ที่ส่งค่าอ่านเป็นระยะ และ data feed (เช่น ข้อมูลตลาดในวงการการเงิน) ล้วนเป็น stream โดยธรรมชาติ เรายังเห็นว่าการมอง write ลงฐานข้อมูลเป็น stream ก็มีประโยชน์ เราจับ changelog — ประวัติของการเปลี่ยนแปลงทั้งหมดที่เกิดกับฐานข้อมูล — ได้ทั้งโดยปริยายผ่าน CDC หรือโดยชัดแจ้งผ่าน event sourcing Log compaction ทำให้ stream เก็บสำเนาเนื้อหาฐานข้อมูลทั้งหมดไว้ได้

การมองฐานข้อมูลเป็น stream เปิดโอกาสที่ทรงพลังสำหรับการ integrate ระบบ คุณทำให้ derived data system เช่น search index, cache และระบบ analytics ทันสมัยอยู่เสมอได้ โดย consume log ของการเปลี่ยนแปลง แล้ว apply กับระบบที่ derive มา คุณยังสร้าง view ใหม่บนข้อมูลที่มีอยู่ได้ด้วย โดยเริ่มจากศูนย์และ consume log ของการเปลี่ยนแปลงตั้งแต่ต้นจนถึงปัจจุบัน

ความสามารถในการเก็บ state เป็น stream และการ replay message ยังเป็นพื้นฐานของเทคนิคที่ทำให้ stream join และ fault tolerance ในหลาย stream processing framework เป็นไปได้ เราได้พูดถึงจุดประสงค์หลายอย่างของ stream processing รวมถึงการค้นหา event pattern (complex event processing) การคำนวณ windowed aggregation (stream analytics) และการทำให้ derived data system ทันสมัยอยู่เสมอ (materialized view)

จากนั้นเราได้พูดถึงความยากของการคิดเรื่องเวลาใน stream processor รวมถึงความต่างระหว่าง processing time กับ event timestamp และปัญหาของ straggler event ที่มาถึงหลังจากที่คุณคิดว่า window เสร็จสมบูรณ์แล้ว

เราแยก join ออกเป็นสามประเภทที่อาจปรากฏใน stream process:

Stream–stream joins

Input stream ทั้งสองประกอบด้วย activity event และ join operator ค้นหา event ที่เกี่ยวข้องกันซึ่งเกิดขึ้นภายใน window ของเวลา ตัวอย่างเช่น join operator อาจจับคู่การกระทำสองอย่าง ที่ผู้ใช้คนเดียวกันทำห่างกันไม่เกิน 30 นาที Join input ทั้งสองอาจเป็น stream เดียวกันจริงๆ (เรียกว่า self join ) ถ้าคุณต้องการหา event ที่เกี่ยวข้องกันภายใน stream นั้น

Stream–table joins

Input stream หนึ่งประกอบด้วย activity event ส่วนอีก stream เป็น database changelog Changelog ช่วยให้สำเนาในเครื่องของฐานข้อมูลทันสมัยอยู่เสมอ สำหรับ activity event แต่ละตัว join operator จะ query ฐานข้อมูล แล้วส่ง activity event ที่ enrich แล้วออกไป

Table–table joins

Input stream ทั้งสองเป็น database changelog ในกรณีนี้ ทุกการเปลี่ยนแปลงที่ฝั่งหนึ่ง จะถูก join กับ state ล่าสุดของอีกฝั่ง ผลลัพธ์คือ stream ของการเปลี่ยนแปลงของ materialized view ของ join ระหว่างสองตาราง

สุดท้าย เราได้พูดถึงเทคนิคสำหรับบรรลุ fault tolerance และ exactly-once semantics ใน stream processor เช่นเดียวกับ batch processing เราต้องทิ้ง output บางส่วนของ task ที่ล้มเหลว แต่เนื่องจาก stream process ทำงานยาวนาน และสร้าง output อย่างต่อเนื่อง เราจึงทิ้ง output ทั้งหมดไปเฉยๆ ไม่ได้ แต่ใช้กลไกการกู้คืนที่ละเอียดกว่าแทนได้ ซึ่งอาศัย microbatching, checkpointing, transaction หรือ idempotent write