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

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 อาจเป็น record ที่มีคีย์ต่ำที่สุดซึ่งต้องเป็น record แรกของ output ดังนั้นการเริ่มสร้าง output ก่อนจึงไม่ใช่ทางเลือก

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

ปัญหาของ daily batch processes คือการเปลี่ยนแปลงใน input จะสะท้อนใน output ช้ากว่าหนึ่งวัน ซึ่งช้าเกินไปสำหรับผู้ใช้ที่ใจร้อนจำนวนมาก เพื่อลดความล่าช้า เราสามารถรันการประมวลผลให้บ่อยขึ้น — เช่น การประมวลผลข้อมูลหนึ่งวินาทีเมื่อสิ้นสุดแต่ละวินาที — หรือแม้กระทั่งทำอย่างต่อเนื่อง โดยไม่แบ่งเวลาเป็นช่วงตายตัว และประมวลผลทุกเหตุการณ์ทันทีที่เกิดขึ้น นี่คือแนวคิดเบื้องหลัง 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) ขั้นตอนแรกโดยปกติคือการแยกวิเคราะห์เป็นลำดับของ records ในบริบทของ stream processing record มักถูกเรียกว่า event (เหตุการณ์) แต่โดยพื้นฐานแล้วมันคือสิ่งเดียวกัน: ออบเจกต์ขนาดเล็กที่เป็นตัวของตัวเองและไม่เปลี่ยนรูป ซึ่งประกอบด้วยรายละเอียดของสิ่งที่เกิดขึ้น ณ จุดเวลาใดเวลาหนึ่ง Event มักจะประกอบด้วย timestamp ที่ระบุว่าเกิดขึ้นเมื่อไหร่ตามนาฬิกาเวลาของวัน (ดู "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

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

อย่างไรก็ตาม เมื่อขยับไปสู่การประมวลผลอย่างต่อเนื่องที่มีความล่าช้าต่ำ การ polling จะมีค่าใช้จ่ายสูง ถ้า datastore ไม่ได้ออกแบบมาสำหรับการใช้งานลักษณะนี้ ยิ่งคุณ poll บ่อยเท่าไหร่ เปอร์เซ็นต์ของ requests ที่ส่งคืน events ใหม่ก็จะยิ่งต่ำลง และค่าใช้จ่าย overhead ก็จะสูงขึ้นตามไปด้วย ดังนั้นจึงดีกว่าถ้า consumers ได้รับการแจ้งเตือนเมื่อมี events ใหม่เกิดขึ้น

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

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

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

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

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

จะเกิดอะไรขึ้นถ้า producers ส่งข้อความเร็วกว่าที่ consumers จะประมวลผลไหว?

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

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

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

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

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

คุณสมบัติที่ดีอย่างหนึ่งของ batch processing systems ที่เราสำรวจใน บทที่ 11 คือพวกมันให้การรับประกันความน่าเชื่อถือที่แข็งแกร่ง งานที่ล้มเหลวจะถูกลองใหม่โดยอัตโนมัติ และ output บางส่วนจากงานที่ล้มเหลวจะถูกยกเลิกโดยอัตโนมัติ นั่นหมายความว่า 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 ต้องจดจำแพ็กเก็ตที่ส่งไปเพื่อให้สามารถส่งซ้ำได้เมื่อมีการร้องขอ)

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

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

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

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

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

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

ทางเลือกที่ใช้กันอย่างแพร่หลายอีกวิธีหนึ่งคือการส่งข้อความผ่าน message broker (หรือที่เรียกว่า message queue ) ซึ่งโดยพื้นฐานแล้วเป็นฐานข้อมูลชนิดหนึ่งที่ถูกปรับให้เหมาะสมสำหรับการจัดการสตรีมข้อความ [ 12 ] มันทำงานเป็นเซิร์ฟเวอร์ โดยมี producers และ consumers เชื่อมต่อเข้ามาเป็น clients Producers เขียนข้อความไปยัง broker และ broker จะส่งข้อความเหล่านั้นไปยัง consumers

โดยการรวมศูนย์ข้อมูลไว้ที่ broker ระบบเหล่านี้สามารถทนต่อ clients ที่มาๆ ไปๆ (เชื่อมต่อ, ตัดการเชื่อมต่อ, และ crash) ได้ง่ายขึ้น และคำถามเรื่อง durability ก็ถูกย้ายไปที่ broker แทน บาง message brokers เก็บข้อความไว้ในหน่วยความจำเท่านั้น ในขณะที่บางตัว (ขึ้นอยู่กับการตั้งค่า) เขียนลงดิสก์เพื่อไม่ให้สูญหายเมื่อ broker crash เมื่อเผชิญกับ consumers ที่ช้า โดยทั่วไปแล้วพวกมันอนุญาตให้ queue แบบไม่จำกัด (แทนที่จะปล่อยข้อความทิ้งหรือใช้ backpressure) แม้ว่าตัวเลือกนี้อาจขึ้นอยู่กับการตั้งค่าด้วยเช่นกัน

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

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

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

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

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

• ฐานข้อมูลมักรองรับ secondary indexes และวิธีการค้นหาข้อมูลต่างๆ ด้วย query language ในขณะที่ message brokers มักรองรับการสมัครรับข้อมูล (subscribe) เฉพาะ subset ของ topics ที่ตรงกับ pattern หนึ่งๆ ทั้งสองวิธีเป็นวิธีให้ client เลือกส่วนของข้อมูลที่ต้องการทราบ แต่ฐานข้อมูลมักมีฟังก์ชันการค้นหาที่ล้ำหน้ากว่ามาก

• เมื่อ query ฐานข้อมูล ผลลัพธ์โดยทั่วไปจะขึ้นอยู่กับ point-in-time snapshot ของข้อมูล ถ้า client อื่นเขียนข้อมูลลงฐานข้อมูลที่เปลี่ยนผลลัพธ์ของ query client แรกจะไม่รู้ว่าผลลัพธ์ก่อนหน้านี้ล้าสมัยแล้ว (เว้นแต่มันจะ query ซ้ำหรือ poll เพื่อดูการเปลี่ยนแปลง) ในทางตรงกันข้าม message brokers ไม่สนับสนุน queries ตามอำเภอใจ และไม่อนุญาตให้อัปเดตข้อความหลังจากถูกส่งไปแล้ว แต่พวกมันจะแจ้งเตือน clients เมื่อข้อมูลเปลี่ยนแปลง (เช่น เมื่อมีข้อความใหม่พร้อมใช้งาน)

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

Multiple consumers (ผู้บริโภคหลายราย)

เมื่อ consumers หลายรายอ่านข้อความใน topic เดียวกัน มีสอง pattern หลักของการรับส่งข้อความ ดังที่แสดงใน Figure 12-1 :

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

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

Fan-out (การกระจายแบบแฟนเอาท์)

แต่ละข้อความถูกส่งไปยัง ทั้งหมด ของ consumers Fan-out อนุญาตให้ consumers อิสระหลายตัวแต่ละตัว "ปรับตัวรับ" เข้ากับ broadcast ของข้อความเดียวกันโดยไม่กระทบกัน — ซึ่งเทียบเท่ากับ streaming ของการมี batch jobs หลายตัวที่อ่านไฟล์ input เดียวกัน (คุณสมบัตินี้มีให้โดย topic subscriptions ใน JMS และ exchange bindings ใน AMQP)

Figure 12-1. (a) Load balancing แบ่งงานในการ consume topic ให้กับ consumers; (b) แบบ fan-out แต่ละข้อความจะถูกส่งไปยัง consumers หลายราย

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

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

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

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

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

Figure 12-2. Consumer 2 crash ขณะประมวลผล m3 ดังนั้นมันจึงถูกส่งซ้ำไปยัง consumer 1 ในเวลาต่อมา

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

การส่งซ้ำยังอาจทำให้เกิดการสิ้นเปลืองทรัพยากร, การขาดแคลนทรัพยากร, หรือการอุดตันถาวรใน stream สถานการณ์ทั่วไปคือ producer ที่ serialize ข้อความไม่ถูกต้อง — เช่น การละเว้น key ที่จำเป็นในออบเจกต์ที่เข้ารหัส JSON ถ้าข้อความที่มี key ขาดหายไปทำให้ consumer crash และรีสตาร์ท มันจะไม่ยอมรับข้อความ ดังนั้น broker จะส่งอีกครั้ง ซึ่งจะทำให้ consumer อื่นล้มเหลว ลูปนี้จะทำซ้ำไปเรื่อยๆ อย่างไม่มีที่สิ้นสุด ถ้า broker รับประกันการจัดลำดับที่เข้มงวด จะไม่มีความคืบหน้าเพิ่มเติมเกิดขึ้นได้ Brokers ที่อนุญาตให้จัดลำดับข้อความใหม่สามารถดำเนินการต่อได้ แต่จะสิ้นเปลืองทรัพยากรกับข้อความที่จะไม่ได้รับการยอมรับ

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

Log-Based Message Brokers (ตัวกลางรับส่งข้อความแบบ Log-Based)

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

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

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

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

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

Using logs for message storage (การใช้ logs สำหรับจัดเก็บข้อความ)

log คือลำดับของ records บนดิสก์ที่ต่อท้ายอย่างเดียว (append-only) เราเคยพูดถึง logs ในบริบทของ log-structured storage engines และ write-ahead logs ใน บทที่ 4 , ในบริบทของการจำลองแบบใน บทที่ 6 , และในรูปแบบของ consensus ใน บทที่ 10

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

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

Figure 12-3. Producers ส่งข้อความโดยการต่อท้ายไปยังไฟล์ topic partition และ consumers อ่านไฟล์เหล่านี้ตามลำดับ

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

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

Logs compared to traditional messaging (การเปรียบเทียบ Logs กับการรับส่งข้อความแบบดั้งเดิม)

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

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

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

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

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

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

Consumer offsets (ตำแหน่งออฟเซตของผู้บริโภค)

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

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

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

Disk space usage (การใช้พื้นที่ดิสก์)

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

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

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

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

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

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

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

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

พฤติกรรมนี้แตกต่างจาก message brokers แบบดั้งเดิม ที่คุณต้องระมัดระวังในการลบ queues ที่ consumers ถูกปิดไปแล้ว เพื่อหลีกเลี่ยงการสะสมข้อความโดยไม่จำเป็นและแย่งหน่วยความจำจาก consumers ที่ทำงานอยู่

Replaying old messages (การเล่นซ้ำข้อความเก่า)

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

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

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

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

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

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

แต่คุณไม่จำเป็นต้องไปไกลถึงขนาดใช้ event sourcing แม้กับ data models ที่ mutable event streams ก็มีประโยชน์สำหรับฐานข้อมูล ในความเป็นจริง ทุก write ไปยังฐานข้อมูลคือ event ที่สามารถถูกจับ, จัดเก็บ, และประมวลผลได้ ความเชื่อมโยงระหว่างฐานข้อมูลและ streams นั้นลึกซึ้งกว่าแค่การจัดเก็บ logs ทางกายภาพบนดิสก์ — มันค่อนข้างเป็นพื้นฐาน

ตัวอย่างเช่น replication log (ดู "Implementation of Replication Logs" ) คือ stream ของ database write events ที่ถูกสร้างโดย leader เมื่อมันประมวลผล transactions Followers นำ stream ของ writes ไปใช้กับสำเนาฐานข้อมูลของตนเอง และจบลงด้วยข้อมูลที่ถูกต้องเหมือนกัน Events ใน replication log อธิบายการเปลี่ยนแปลงข้อมูลที่เกิดขึ้น

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

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

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

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

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

ถ้า periodic full database dumps ช้าเกินไป ทางเลือกที่บางครั้งใช้คือ dual writes (การเขียนสองทาง) ซึ่งโค้ดแอปพลิเคชันเขียนไปยังแต่ละระบบอย่างชัดเจนเมื่อข้อมูลเปลี่ยนแปลง — ตัวอย่างเช่น เขียนไปยังฐานข้อมูลก่อน, จากนั้นอัปเดต search index, แล้วทำให้ cache entries หมดอายุ (หรือแม้กระทั่งทำ writes เหล่านั้นพร้อมกัน)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

คุณสมบัติ log compaction นี้รองรับโดย Apache Kafka ดังที่เราจะได้เห็นต่อไปในบทนี้ มันช่วยให้ message broker สามารถใช้สำหรับการจัดเก็บแบบ durable ไม่ใช่แค่การรับส่งข้อความชั่วคราวเท่านั้น

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

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

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

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

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

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

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

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

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

Change Data Capture and Database Schemas (Change Data Capture และ Schemas ของฐานข้อมูล)

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

อย่างไรก็ตาม ระบบ CDC โดยทั่วไปใช้ schema ของ upstream database เมื่อทำซ้ำข้อมูล ซึ่งเปลี่ยน schemas เหล่านี้ให้เป็น public APIs ที่ต้องจัดการเช่นเดียวกับ public API ของ service การลบคอลัมน์ในตารางฐานข้อมูลจะทำลาย downstream consumers ที่ขึ้นอยู่กับฟิลด์นั้น ความท้าทายดังกล่าวมีอยู่เสมอกับ data pipelines แต่โดยปกติมันส่งผลกระทบเฉพาะ data warehouse ETL เนื่องจาก CDC มักถูก implement เป็น data stream services production อื่นอาจเป็น consumers การทำลาย consumers ดังกล่าวอาจทำให้เกิดการหยุดชะงักที่กระทบลูกค้า [ 34 ] Data contracts มักถูกใช้เพื่อป้องกันการเสียหายเหล่านี้

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

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

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

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

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

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

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

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

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

เมื่อใดก็ตามที่คุณมีสถานะที่เปลี่ยนแปลง สถานะนั้นเป็นผลลัพธ์ของ events ที่กลายพันธุ์มันเมื่อเวลาผ่านไป ตัวอย่างเช่น รายชื่อที่นั่งว่างในปัจจุบันคือผลลัพธ์ของการจองที่คุณได้ดำเนินการ ยอดเงินในบัญชีปัจจุบันคือผลลัพธ์ของเครดิตและเดบิตในบัญชี และกราฟ response time สำหรับ web server ของคุณคือการรวมของ response time แต่ละรายการของ web requests ทั้งหมดที่เกิดขึ้น

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

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

Figure 12-7. ความสัมพันธ์ระหว่างสถานะแอปพลิเคชันปัจจุบันและ event stream

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

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

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

Advantages of immutable events (ข้อดีของ Events ที่ไม่เปลี่ยนรูป)

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

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

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

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

Deriving several views from the same event log (การสร้างมุมมองหลายมุมจาก Event Log เดียวกัน)

โดยการแยก mutable state ออกจาก immutable event log คุณสามารถสร้าง read-oriented representations หลายแบบ จาก log ของ events เดียวกัน สิ่งนี้ทำงานเหมือนกับการมี consumers หลายตัวของ stream (Figure 12-5 ) — ตัวอย่างเช่น analytical database Druid ingest โดยตรงจาก Kafka โดยใช้แนวทางนี้ และ Kafka Connect sinks สามารถส่งออกข้อมูลจาก Kafka ไปยังฐานข้อมูลและ indexes ต่างๆ [ 33 ]

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

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

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

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

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

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

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

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

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

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

Limitations of immutability (ข้อจำกัดของความไม่เปลี่ยนรูป)

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

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

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

การลบข้อมูลจริงๆ นั้นยากอย่างน่าประหลาดใจ [ 49 ] เพราะสำเนาสามารถอยู่ในหลายที่ Storage engines, filesystems, และ SSDs มักเขียนไปยังตำแหน่งใหม่แทนที่จะเขียนทับข้อมูลในที่เดิม [ 41 ] และ backups มักจะถูกทำให้ immutable โดยเจตนา เพื่อป้องกันการลบหรือเสียหายโดยไม่ได้ตั้งใจ

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

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

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

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

จนถึงตอนนี้ในบทนี้เราได้พูดถึงว่า streams มาจากไหน (กิจกรรมผู้ใช้, เซนเซอร์, และ writes ไปยังฐานข้อมูล) และ streams ถูกส่งอย่างไร (ผ่าน direct messaging, ผ่าน message brokers, และใน event logs)

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

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

2. คุณสามารถส่ง events ไปยังผู้ใช้ในบางวิธี — ตัวอย่างเช่น โดยการส่ง email alerts หรือ push notifications หรือโดยการ streaming events ไปยัง real-time dashboard ที่พวกมันถูกแสดงเป็นภาพ ในกรณีนี้ มนุษย์เป็นผู้บริโภคสุดท้ายของ stream

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

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

Patterns สำหรับ sharding และ parallelization ใน stream processors ก็คล้ายกันมากกับ MapReduce และ dataflow engines ที่เราเห็นใน บทที่ 11 ดังนั้นเราจะไม่พูดซ้ำหัวข้อเหล่านี้ Basic mapping operations เช่น การแปลงและการกรอง records ก็ทำงานเหมือนกัน

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

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

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

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

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

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

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

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

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

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

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

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

การ implement ของ CEP รวมถึง Esper, Apama, และ TIBCO StreamBase Distributed stream processors อย่าง Flink และ Spark Streaming ก็รองรับ SQL สำหรับ declarative queries บน streams เช่นกัน

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

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

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

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

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

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

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

กรอบงาน stream processing แบบโอเพนซอร์สที่มีชื่อเสียงหลายตัว เช่น Apache Storm, Spark Streaming, Flink, Samza, Apache Beam, และ Kafka Streams ถูกออกแบบมาเพื่อการวิเคราะห์ [ 57 ] บริการแบบ hosted รวมถึง Google Cloud Dataflow และ Azure Stream Analytics

Maintaining materialized views (การบำรุงรักษา Materialized Views)

เราเห็นว่า stream ของการเปลี่ยนแปลงไปยังฐานข้อมูล สามารถใช้เพื่อให้ derived data systems เช่น caches, search indexes, และ data warehouses อัปเดตอยู่เสมอด้วย source database เหล่านี้คือตัวอย่างของการบำรุงรักษา materialized views: การสร้างมุมมองทางเลือกของชุดข้อมูลเพื่อให้คุณสามารถ query ได้อย่างมีประสิทธิภาพ และอัปเดตมุมมองนั้นเมื่อใดก็ตามที่ข้อมูลพื้นฐานเปลี่ยนแปลง [ 37 ]

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

โดยหลักการแล้ว stream processor ใดๆ ก็สามารถใช้ สำหรับการบำรุงรักษา materialized view ได้ แม้ว่าความจำเป็นในการเก็บ events ตลอดไป จะขัดแย้งกับข้อสมมติของกรอบงานที่เน้น analytics บางตัว ที่ส่วนใหญ่ทำงานบน windows ที่มีระยะเวลาจำกัด Kafka Streams และ ksqlDB ของ Confluent รองรับการใช้งานแบบนี้ โดยสร้างบนการสนับสนุน log compaction ของ Kafka [ 58 ]

Incremental View Maintenance (การบำรุงรักษามุมมองแบบเพิ่มทีละ)

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

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

ประสิทธิภาพต่ำ

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

ความสดใหม่ของข้อมูล

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

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

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

ฐานข้อมูลอย่าง Materialize [ 61 ], RisingWave, ClickHouse, และ Feldera ทั้งหมดใช้เทคนิค IVM เพื่อให้ materialized views แบบ incremental ที่มีประสิทธิภาพ ฐานข้อมูลเหล่านี้ ingest streams ของ events เพื่อเปิดเผย materialized views แบบ real-time Events ล่าสุดถูก buffer ในหน่วยความจำ และใช้เพื่ออัปเดต materialized views บนดิสก์เป็นระยะ การอ่านรวม events ล่าสุดและข้อมูลที่ถูก materialized เพื่อให้มุมมอง real-time เดียว เนื่องจากการอ่านมักแสดงใน SQL และ materialized views มักถูกจัดเก็บในรูปแบบ OLAP-style ระบบเหล่านี้ยังรองรับ queries แบบ data warehouse ขนาดใหญ่ เช่นที่กล่าวถึงใน บทที่ 11

Search on streams (การค้นหาบนสตรีม)

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

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

Event-driven architectures and RPC (สถาปัตยกรรมที่ขับเคลื่อนด้วยเหตุการณ์และ RPC)

ในหัวข้อ "Event-Driven Architectures" เราได้พูดถึงระบบส่งข้อความเป็นทางเลือกแทน RPC กลไกนี้สำหรับ services ในการสื่อสารถูกใช้ ตัวอย่างเช่น ใน actor model

แม้ว่าระบบเหล่านี้จะโดยอาศัยข้อความและ events แต่โดยปกติเราไม่ได้คิดถึงพวกมันในฐานะ stream processors ด้วยเหตุผลบางประการ:

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

• การสื่อสารระหว่าง actors มักเป็น ephemeral และ one-to-one ในขณะที่ event logs เป็น durable และ multi-subscriber

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

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

นอกจากนี้ยังเป็นไปได้ที่จะประมวลผล streams โดยใช้ actor frameworks อย่างไรก็ตาม กรอบงานดังกล่าวจำนวนมากไม่รับประกันการส่งข้อความ ในกรณีที่เกิด crashes ดังนั้นการประมวลผลจึงไม่ fault-tolerant เว้นแต่คุณจะ implement ตรรกะการลองซ้ำเพิ่มเติม

Reasoning About Time (การให้เหตุผลเกี่ยวกับเวลา)

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

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

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

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

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

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

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

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

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

Figure 12-8. Windowing ตาม processing time ทำให้เกิดสิ่งแปลกปลอม เนื่องจากการเปลี่ยนแปลงของอัตราการประมวลผล

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

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

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

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

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

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

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

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

การกำหนด timestamps ให้กับ events นั้นยากยิ่งขึ้น เมื่อ events สามารถถูก buffer ณ จุดต่างๆ ในระบบ ตัวอย่างเช่น พิจารณาแอปมือถือที่รายงาน events สำหรับ metrics การใช้งานไปยังเซิร์ฟเวอร์ แอปอาจถูกใช้ขณะที่อุปกรณ์ออฟไลน์ ซึ่งในกรณีนี้มันจะ buffer events ในเครื่อง และส่งพวกมันไปยังเซิร์ฟเวอร์เมื่อมีการเชื่อมต่ออินเทอร์เน็ตครั้งถัดไป (ซึ่งอาจเป็นชั่วโมงหรือหลายวันต่อมา) สำหรับ consumers ใดๆ ของ stream นี้ events จะปรากฏเป็น stragglers ที่ล่าช้าอย่างมาก

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

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

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

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

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

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

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

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

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

Tumbling windows (หน้าต่างแบบต่อเนื่องไม่ซ้อนทับ)

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

Hopping windows (หน้าต่างแบบกระโดด)

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

Sliding windows (หน้าต่างแบบเลื่อน)

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

Session windows (หน้าต่างแบบเซสชัน)

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

การดำเนินการแบบ window มักจะรักษาสถานะชั่วคราว ในบางกรณี สถานะมีขนาดคงที่ ไม่ว่า window จะใหญ่แค่ไหนหรือมี events กี่รายการ — ตัวอย่างเช่น การนับจะมี counter ตัวเดียว โดยไม่ขึ้นกับขนาด window หรือจำนวน events ในทางกลับกัน sliding windows หรือ stream joins ซึ่งเราจะพูดถึงในส่วนถัดไป ต้องการให้ events ถูก buffer จนกว่า window จะเสร็จสิ้น ดังนั้นขนาด window ใหญ่หรือ streams ที่มี throughput สูง อาจทำให้ stream processors ต้องเก็บสถานะชั่วคราวจำนวนมาก คุณต้องแน่ใจว่าเครื่องที่รันงาน stream processing มีความจุเพียงพอที่จะรักษาสถานะนี้ ไม่ว่าจะในหน่วยความจำหรือบนดิสก์

Stream Joins (การ Join สตรีม)

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

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

Stream–stream join (window join) (การ Join สตรีมกับสตรีม)

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

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

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

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

Stream–table join (stream enrichment) (การ Join สตรีมกับตาราง)

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

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

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

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

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

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

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

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

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

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

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

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

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

อีกวิธีในการมอง stream process นี้คือ มันรักษา 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 ของ streams สอดคล้องโดยตรงกับ join ของตารางใน query นี้ Timelines เป็น cache ของผลลัพธ์ของ query อย่างมีประสิทธิภาพ ที่อัปเดตทุกครั้งที่ตารางพื้นฐานเปลี่ยนแปลง

Note

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

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

joins ทั้งสามประเภทที่อธิบายที่นี่ (stream–stream, stream–table, และ table–table) มีอะไรที่เหมือนกันมาก พวกมันทั้งหมดต้องการให้ stream processor รักษาสถานะ (search และ click events, user profiles, หรือ follower list) ที่ derived จาก join input หนึ่ง และ query สถานะนั้นเมื่อประมวลผล records จาก input อื่น

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

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

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

ถ้าลำดับของ events ข้าม streams ไม่ถูกกำหนด join จะกลายเป็น nondeterministic [ 70 ] ซึ่งหมายความว่าคุณไม่สามารถรัน job เดียวกัน บน input เดียวกันและได้รับผลลัพธ์เดียวกันเสมอ Events บน input streams อาจถูกสลับกันในวิธีที่แตกต่างกัน เมื่อคุณรัน job อีกครั้ง

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

Fault Tolerance (ความทนทานต่อข้อผิดพลาด)

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

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

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

Microbatching and checkpointing (การแบ่งเป็นชุดย่อยและการตรวจสอบจุด)

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

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

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

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

Atomic commit revisited (การคอมมิตแบบอะตอมมิกทบทวนอีกครั้ง)

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

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

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

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

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

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

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

การพึ่งพา idempotence หมายถึงข้อสมมติหลายประการ: การ restart งานที่ล้มเหลวต้องเล่นข้อความเดียวกันซ้ำในลำดับเดียวกัน (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 (การสร้างสถานะใหม่หลังจากข้อผิดพลาด)

กระบวนการ stream ใดๆ ที่ต้องใช้สถานะ — เช่น windowed aggregations (เช่น counters, averages, และ histograms) และ tables และ indexes ใดๆ ที่ใช้สำหรับ joins — ต้องแน่ใจว่าสถานะนี้สามารถกู้คืนได้หลังจากข้อผิดพลาด

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

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

ในบางกรณี การจำลองแบบสถานะอาจไม่จำเป็นด้วยซ้ำ เพราะมันสามารถสร้างขึ้นใหม่จาก input streams ตัวอย่างเช่น ถ้าสถานะประกอบด้วยการรวมของ window ที่ค่อนข้างสั้น มันอาจเร็วพอที่จะเล่น input events ที่สอดคล้องกับ window นั้นซ้ำ ถ้าสถานะเป็น replica ท้องถิ่นของฐานข้อมูล ที่บำรุงรักษาโดย CDC ฐานข้อมูลก็สามารถสร้างขึ้นใหม่จาก change stream ที่ถูก log-compacted

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

Summary (สรุป)

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

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

AMQP/JMS-style message broker

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

Log-based message broker

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

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

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

การแสดงฐานข้อมูลเป็น streams เปิดโอกาสที่มีประสิทธิภาพสำหรับการรวมระบบ คุณสามารถทำให้ derived data systems เช่น search indexes, caches, และระบบวิเคราะห์ อัปเดตอยู่เสมอโดยการ consume log ของการเปลี่ยนแปลง และนำไปใช้กับระบบที่ถูก derived คุณยังสามารถสร้างมุมมองใหม่บนข้อมูลที่มีอยู่ โดยเริ่มจากศูนย์และ consume log ของการเปลี่ยนแปลงตั้งแต่ต้นจนถึงปัจจุบัน

สิ่งอำนวยความสะดวกสำหรับการรักษาสถานะเป็น streams และการเล่นซ้ำข้อความ ยังเป็นพื้นฐานสำหรับเทคนิคที่ทำให้ stream joins และ fault tolerance ในกรอบงาน stream processing ต่างๆ เราได้พูดถึงวัตถุประสงค์หลายประการของ stream processing รวมถึงการค้นหารูปแบบเหตุการณ์ (complex event processing), การคำนวณ windowed aggregations (stream analytics), และการทำให้ระบบ derived data อัปเดตอยู่เสมอ (materialized views)

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

เราแยกความแตกต่างของ joins สามประเภทที่อาจปรากฏใน stream processes:

Stream–stream joins

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

Stream–table joins

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

Table–table joins

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

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