Data Pipeline ของคุณยุ่งวุ่นวาย นี่คือวิธีหยุดไม่ให้ข้อมูลซ้ำซ้อนสูญเปล่าเป็นล้าน

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

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

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

มาดูสามแนวทางหลักในการกำจัดข้อมูลซ้ำซ้อนในระบบสตรีมข้อมูลกัน:

การกำจัดข้อมูลซ้ำซ้อนโดยใช้แอตทริบิวต์ข้อความ Pub/Sub

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

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

มาพูดคุยกันในเชิงเทคนิค Google Cloud ให้บริการ Pub/Sub ซึ่งเป็นบริการส่งข้อความแบบอะซิงโครนัสและปรับขนาดได้ โดยแยกบริการที่สร้างข้อความออกจากบริการที่ประมวลผลข้อความเหล่านั้น โดยบริการนี้ใช้กันอย่างแพร่หลายสำหรับการวิเคราะห์แบบสตรีมและการรวมข้อมูลเพื่อโหลดและแจกจ่ายข้อมูล โดยส่วนใหญ่มักใช้สำหรับการรับเหตุการณ์การโต้ตอบของผู้ใช้ เหตุการณ์เซิร์ฟเวอร์ เหตุการณ์แบบเรียลไทม์ การจำลองข้อมูลระหว่างฐานข้อมูล ทำหน้าที่เป็นบัสเหตุการณ์ขององค์กรเพื่อแบ่งปันเหตุการณ์ทางธุรกิจในองค์กร และสตรีมข้อมูลจากแอปพลิเคชันต่างๆ รวมถึงเซ็นเซอร์และเหตุการณ์แอปพลิเคชันที่ใช้ร่วมกับผลิตภัณฑ์ Google Cloud อื่นๆ ผ่านข้อมูลไปป์ไลน์

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

สำหรับข้อความที่มีฟิลด์ที่ไม่ซ้ำกัน เช่น transaction_id คุณสามารถตั้งค่าค่านี้เป็นแอตทริบิวต์ได้เมื่อเผยแพร่ข้อความ เมื่ออ่านข้อความจาก Pub/Sub ใน Dataflow คุณสามารถกำหนดค่าไปป์ไลน์เพื่อขจัดข้อมูลซ้ำซ้อนโดยใช้แอตทริบิวต์นี้

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

อาจมีบางกรณีที่ข้อมูลซ้ำกันที่สร้างขึ้นภายในตัวเผยแพร่เอง เช่น Pub/Sub เนื่องมาจากความล่าช้าในการใช้ข้อความโดยดาวน์สตรีม หรือ Pub/Sub ไม่เคยได้รับการยืนยันสำหรับข้อความที่ส่ง Pub/Sub พยายามส่งข้อความเดียวกันอีกครั้งโดยใช้ Message_id เดียวกัน จึงสร้างเหตุการณ์ซ้ำกันในตัวเผยแพร่ เพื่อแก้ปัญหานี้ ให้ใช้ Pub/Sub เพื่อกำหนด message_id ของเพย์โหลดและใช้สิ่งนี้เป็นตัวระบุ Cloud DataFlow เป็นบริการที่มีการจัดการอย่างสมบูรณ์สำหรับการประมวลผลข้อมูลแบบสตรีมบน แพลตฟอร์ม Google Cloud (GCP) ซึ่งให้การประมวลผลระเบียนแต่ละรายการ เพียงครั้งเดียว สิ่งนี้มีความหมายสำหรับเราอย่างไร - บริการนี้จะระบุเหตุการณ์ซ้ำกันตาม message_id และกำจัดเหตุการณ์เหล่านี้เมื่อประมวลผลในไพพ์ไลน์ข้อมูล แต่ในบางกรณี เหตุการณ์ซ้ำกันเหล่านี้เมื่อประมวลผลบนโหนดเวิร์กเกอร์ที่แตกต่างกันภายในกระแสข้อมูล เหตุการณ์เหล่านี้จะถูกส่งไปยังดาวน์สตรีมอย่างไม่มีประสิทธิภาพ คุณจะยังคงมีเหตุการณ์ซ้ำกันในดาต้าเลคของคุณ

เราจะหารือเพิ่มเติมเกี่ยวกับวิธีการจัดการกรณีดังกล่าวในตอนท้ายบทความนี้ มาดูตัวเลือกที่เหลือในการกำจัดข้อมูลซ้ำซ้อนของสตรีมกันดีกว่า

การกำจัดข้อมูลซ้ำซ้อนโดยใช้ Deduplicate PTransform ของ Apache Beam

ตอนนี้เราทราบแล้วว่า Pub/Sub จัดการเหตุการณ์ซ้ำซ้อนอย่างไร ขั้นตอนต่อไปคือการประมวลผลข้อความเหล่านี้โดยใช้ Cloud DataFlow โดยที่ผู้สมัคร Pub/Sub อ่านข้อความสตรีมจากแอปพลิเคชันต้นทาง Dataflow เป็นบริการที่ได้รับการจัดการอย่างสมบูรณ์ซึ่งใช้ Apache Beam SDK แบบโอเพ่นซอร์สเพื่อเปิดใช้งานความสามารถในการสตรีมขั้นสูง Dataflow ขยายขนาดได้ถึง 4,000 โหนดเวิร์กเกอร์ต่องาน และสามารถประมวลผลข้อมูลได้หลายเพตาไบต์ด้วยคุณสมบัติการปรับขนาดอัตโนมัติเพื่อใช้ทรัพยากรได้ดีขึ้นทั้งในไปป์ไลน์แบบแบตช์และแบบสตรีม

Apache Beam นำเสนอ Deduplicate PTransform ในตัวซึ่งให้วิธีการที่กำหนดค่าได้และแข็งแกร่งยิ่งขึ้นสำหรับการลบข้อมูลซ้ำ วิธีนี้ใช้ Stateful API ของ Beam เพื่อรักษาสถานะสำหรับแต่ละคีย์ที่สังเกตและลบข้อมูลซ้ำภายในกรอบเวลาที่ผู้ใช้กำหนด วิธีนี้ช่วยให้คุณกำหนดตรรกะการลบข้อมูลซ้ำตามฟิลด์เฉพาะในข้อมูลของคุณหรือเนื้อหาข้อความทั้งหมด โดยมีความสามารถในการกำหนดค่าการลบข้อมูลซ้ำตาม เวลาของเหตุการณ์ หรือ เวลาในการประมวลผล

ลองดูโค้ดตัวอย่างข้อมูลไปป์ไลน์ของฉันจาก GitHub เพื่อลองใช้ฟังก์ชันนี้

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

โซลูชันนี้ให้ตัวเลือกที่ยืดหยุ่นสำหรับการประมวลผลการเข้าสู่ระบบการลบข้อมูลซ้ำซ้อนที่ซับซ้อน และเหมาะสำหรับสถานการณ์ที่หน้าต่างการลบข้อมูลซ้ำซ้อนมีขนาดใหญ่หรือซับซ้อนกว่าที่ Pub/Sub เสนอ การแลกเปลี่ยนรวมถึงการใช้ทรัพยากรที่มากขึ้นสำหรับการบำรุงรักษาแต่ละสถานะเพื่อกำหนดความไม่ซ้ำกันของระเบียน

การกำจัดข้อมูลซ้ำซ้อนในซิงก์

จนถึงตอนนี้ เราได้เห็นแล้วว่า Publisher เช่น Pub/Sub และบริการการรวมระบบ Cloud DataFlow จัดการกับข้อมูลซ้ำซ้อนแบบเรียลไทม์ได้อย่างไร ฉันคิดว่าโซลูชันเหล่านี้ไม่ได้ผล 100% เมื่อต้องใช้การแบ่งหน้าต่างเนื่องจากปัญหาโอเวอร์เฮดในการประมวลผลและปริมาณข้อมูล ในสถานการณ์เช่นนี้ เพื่อใช้จัดการกรณีขอบ เช่น เมื่อข้อความซ้ำกันมาถึงช้าและดาต้าโฟลว์คิดว่าเป็นเรกคอร์ดเฉพาะเนื่องจากไม่มี ID เรกคอร์ดเพื่อตรวจสอบความซ้ำซ้อนของข้อความ และในสถานการณ์อื่น ดาต้าโฟลว์จัดการข้อความเหล่านี้บนโหนดเวิร์กเกอร์ที่แตกต่างกันเนื่องจากความล้มเหลวของเครือข่ายและ/หรือความล้มเหลวของโหนดเวิร์กเกอร์ทำให้คิดว่าเป็นเรกคอร์ดเฉพาะในขณะที่กำลังประมวลผลในดาต้าโฟลว์และเข้าสู่ระบบดาวน์สตรีม เช่น ตาราง Google Cloud BigQuery

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

ขึ้นอยู่กับกรณีการใช้งาน มีโซลูชันสองประเภทให้เลือกในการแก้ไขข้อมูลที่ซ้ำกัน

ขั้นแรก ให้ใช้แบบสอบถามตามกำหนดเวลาโดยใช้ DAG ของคอมโพเซอร์หรือภายในคอนโซล BigQuery เพื่อสร้างตารางการลบข้อมูลซ้ำซ้อนเป็นระยะๆ โดยใช้พาร์ติชัน (แบบรายวันหรือรายชั่วโมง) ทำให้เป็นตัวเลือกที่ง่ายสำหรับทุกคนในการสร้างกระบวนการและจัดเก็บข้อมูลการลบข้อมูลซ้ำซ้อนในตารางสเตจจิ้ง และโหลดข้อมูลที่แตกต่างกันลงในตารางสุดท้าย

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

นำเสนอแบบสอบถาม Bigquery SQL บนลิงก์ dedup_sql บน Github ของฉัน

ด้านล่างนี้คือโค้ด SQL ของ BigQuery ที่อธิบายตัวเลือกสองตัวที่เราได้กล่าวถึง:

 -- Use below SQL queries to periodically deduplicate data in BigQuery tables. CREATE OR REPLACE TABLE Transactions AS SELECT DISTINCT * FROM raw_transactions; --OR use below incremental steps to drop the necessary partitions and re-insert the deduped data into the original table -- Step 1: Insert distinct records from the original table based on the max timestamp available CREATE OR REPLACE TABLE STAGE_TRANSACTIONS AS SELECT DISTINCT * FROM raw_transactions WHERE event_timestamp > ( SELECT MAX(event_timestamp) FROM raw_transactions ); -- Step 2: Drop the partition after deduplication DELETE FROM raw_transactions WHERE event_timestamp = > ( SELECT MAX(event_timestamp) FROM raw_transactions ); -- Step 3: Insert deduplicated data into the main table INSERT INTO raw_transactions SELECT DISTINCT * FROM STAGE_TRANSACTIONS; --OR Use below SQL query to Merge new data without duplicates the table MERGE INTO raw_transactions AS target USING ( SELECT * FROM STAGE_TRANSACTIONS ) AS source ON target.transaction_id = source.transaction_id AND target.event_timestamp <= source.event_timestamp WHEN MATCHED THEN UPDATE SET target.product = source.product, target.price = source.price, target.location = source.location, target.store = source.store, target.zipcode = source.zipcode, target.city = source.city, target.promotion = source.promotion, target.event_timestamp = source.event_timestamp WHEN NOT MATCHED THEN INSERT (transaction_id, product, price, location, store, zipcode, city, promotion, event_timestamp) VALUES (source.transaction_id, source.product, source.price, source.location, source.store, source.zipcode, source.city, source.promotion, source.event_timestamp); --OR to get the real-time data without duplicates, use following materialized view and a finest solution to retrieve dedup records quickly CREATE MATERIALIZED VIEW raw_transactions_mv AS SELECT transaction_id, product, price, location, store, zipcode, city, promotion, event_timestamp FROM ( SELECT transaction_id, product, price, location, store, zipcode, city, promotion, event_timestamp, ROW_NUMBER() OVER ( PARTITION BY transaction_id ORDER BY event_timestamp DESC ) AS row_num FROM raw_transactions ) WHERE row_num = 1;

การแลกเปลี่ยนทางเทคนิค

กลยุทธ์การกำจัดข้อมูลซ้ำซ้อนแต่ละกลยุทธ์มีทางเลือกที่แตกต่างกัน นี่คือสรุปที่จะช่วยให้คุณเลือกแนวทางที่เหมาะสมได้:

โดยสรุป

การกำจัดข้อมูลซ้ำซ้อนเป็นรากฐานสำคัญของการประมวลผลข้อมูลที่มีประสิทธิภาพในไปป์ไลน์การสตรีม การเลือกกลยุทธ์ขึ้นอยู่กับความต้องการแบบเรียลไทม์ ความซับซ้อน และข้อจำกัดด้านทรัพยากรของไปป์ไลน์ของคุณ ด้วยการใช้ประโยชน์จากจุดแข็งของแอตทริบิวต์ Pub/Sub, Apache Beam Deduplicate PTransform หรือการกำจัดข้อมูลซ้ำซ้อนตามซิงก์ คุณสามารถมั่นใจได้ว่าข้อมูลจะสะอาดและเชื่อถือได้สำหรับระบบดาวน์สตรีม ลองสำรวจแนวทางเหล่านี้ นำตัวอย่างที่ให้มาไปใช้ และปรับให้เข้ากับกรณีการใช้งานของคุณเพื่อให้ได้ผลลัพธ์ที่ดีที่สุด

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