Building Data Lakehouse: Apache Hudi คืออะไร ทำความเข้าใจ Hudi กัน

เนื่องจากที่เราเคยเขียนเกี่ยวกับเรื่อง เข้าใจ Data Warehouse, Data Lake และ Data Lakehouse ฉบับมือใหม่ ที่พูดถึงเบสิคความเข้าใจว่า Data Lake, Data Warehouse และ Data Lakehouse คืออะไรกันไปแล้ว

บทความนี้เราจะมาลงลึกถึงการทำ Data Lakehouse ผ่าน Apache Hudi กัน

สร้าง Data Lakehouse ได้ยังไงบ้าง

เราขอแบ่งออกเป็น 2 วิธีหลักๆ

1. แบบ Managed service
   วิธีนี้เป็นแบบเสียตังแบบไม่ต้องลงแรง build เยอะ เช่น

• Databricks Lakehouse Platform
• Azure Synapse Analytics โดยมี feature ของ Azure Databricks พ่วงมาด้วย

2. แบบ Open source
ฝั่ง open source ที่มักจะมีตัวที่แข่งกันหลักๆ 3 ตัวคือ

• Delta Lake โดย Databricks
• Apache Iceberg โดย Netflix
• Apache Hudi โดย Uber

แต่ไม่ใช่ว่า open source จะไม่เสียตังนะ ก็เสียค่า self-hosted เหมือนกัน

โดยทาง Data Engineer ของ TILDI เองได้เลือก Apache Hudi (อ่านว่า hoodie)

Apache Hudi คืออะไร

Hudi มาจากคำว่า Hadoop Upsert Delete and Incremental

Apache Hudi คือ streaming data lake platform ที่เป็น server-less, transaction layer โดยทำงานอยู่บน HDFS หรือ Cloud Storage platform ต่างๆไม่ว่าจะเป็น Amazon S3, Google Cloud Storage, Azure Blob Storage, Azure Data Lake Gen 2 และอื่นๆ

Hudi ถูกสร้างมาเพื่อ support การรับ streaming ingest และ ingest ข้อมูลแบบ mini-batches

diagram: Building a Large-scale Transactional Data Lake at Uber Using Apache Hudi

ซึ่งความตั้งใจของทาง Uber คืออยากสร้างสิ่งที่ช่วยทำให้การทำ data ingestion และ data preparation ใน Hadoop Distributed File System (HDFS) มัน efficient and low latency มากที่สุด เพื่อมาใช้แก้ปัญหาใหญ่ในการทำ data pipeline ในองค์กร

ดังนั้นเอง Hudi จึง integrate เข้ากับ Hadoop ecosystem ได้ดี ไม่ว่าจะ Apache Hive, Apache Parquet, Presto และ Apache Spark

ที่มาของ Hudi

Uber’s pipeline types. Diagram by Vinoth Chandar

ภาพนี้เป็นภาพที่น่าสนใจมาก ทาง Uber engineer ได้แบ่ง pipeline ออกเป็นรูปแบบต่างๆตาม use-cases

ในเคสของ Uber trip ในแอพ Uber เขาต้องการข้อมูลที่สดใหม่ในการเพิ่ม UX ให้กับทั้ง rider และ driver ตั้งแต่ rider เริ่ม request trip, ระหว่างขึ้นรถ, จนเมื่อ rider ถึง destination ซึ่งทั้ง trip ต้องใช้การ update ข้อมูลที่บ่อยมากๆเพื่อให้ข้อมูลสดใหม่มาโชว์เสมอ

โดยเมื่อก่อน Uber ใช้ Apache Spark ในการ rewrite dataset ใน HDFS ทั้งไฟล์ ซึ่งมีการ insert, update, delete ตลอดเวลา ซึ่งกลายเป็นต้อง process ข้อมูลที่ใหญ่มากๆ เขาเลยเริ่มหาทางออก ครั้น ณ ขณะนั้นก็ยังไม่มี open source tool ตัวไหนที่ตรงกับโจทย์ ดังนั้น Uber ก็เลยสร้างขึ้นมาซะเอง จนในที่สุด Uber ก็ได้โดเนท Hudi ให้กับ Apache Software foundation จนออกมาเป็น open source

ทั้งนี้ เขาต้องทำ batch processing ที่ใช้ computation ที่เยอะมากๆและต้องการข้อมูลแบบ near real-time ไปพร้อมๆกัน จึงเป็นจุดที่ Hudi เกิดมาเพื่อแก้ปัญหานี้โดยเฉพาะ ซึ่งก็คือ Incremental processing

ความแตกต่างระหว่าง Near real-time และ Batch, Stream processing. Source: Vinoth Chandar.

Incremental processing คืออะไร

Incremental processing คือมันจะ process เฉพาะข้อมูลชุดใหม่ที่เข้ามาเท่านั้น โดยไม่ process ทั้งชุดข้อมูล

ซึ่งวิธีการนี้เป็นการเพิ่ม efficiency โดยลดงานที่ต้อง re-processing

ในขณะที่ streaming processing pipeline จะเน้นเรื่อง row-oriented ในการส่งข้อมูลระดับวินาทีและมี latency ที่ต่ำ ตัว incremental pipeline นั้นลอกเลียนคอนเซปนั้นเหมือนกันแต่เน้นใช้กับ columnar data ใน data lake แทน

Diagram by Vinoth Chandar

ทำไมเราถึงเลือกใช้ Hudi

ในช่วงเวลาที่ทีม Data Eng ของ TILDI มีแผนในการปรับ pipeline ต่างๆให้สอดคล้องกับคอนเซปเรื่อง Data Lakehouse

ณ ขณะนั้น มี use case ที่เป็น streaming ingestion อย่าง Kafka เข้ามาพอดี เลยเป็นสาเหตุที่ทำให้ Hudi เป็นผู้เล่นที่น่าสนใจมากกว่า Iceberg ด้วยสาเหตุที่ Uber เขาเคลมว่า

Hudi is a Spark library that is intended to be run as a streaming ingest job, and ingests data as mini-batches

และทีมของเราก็ใช้ Spark เป็นตัวหลักในการ process ข้อมูลอยู่แล้วด้วย พวกเราจึงตัดสินใจเลือกใช้ Hudi ในการ implement Lakehouse นี้

ส่วนประกอบของ Hudi

แน่นอนว่าเราได้กล่าวไปว่า Data Lakehouse คือการ enable เรื่อง ACID guarantee ให้กับ Data Lake ซึ่งตัว Hudi เองก็สามารถทำได้เช่นกัน

Hudi นั้นดังเรื่องในการทำ Upsert ข้อมูล และ Incremental pull เพื่อให้ user ได้ใช้ change data capture กับ Data Lake มากที่สุด

Hudi นั้นแบ่งออกเป็น 3 ส่วนหลัก

• ส่วนการเก็บ Raw data
• ส่วนจัดการ Data Index — เพิ่มความสามารถให้ Upsert เร็วขึ้น
• ส่วนของ Metadata — ในการ manage dataset

หน้าตาตัวอย่างของ Hudi จะเห็นว่ามี field ตัว commit time, record key, partition

ซึ่งนอกจากนี้ Hudi เองยังมีฟีเจอร์ต่างๆที่น่าสนใจไม่ว่าจะเป็น

• Copy On Write และ Merge On Read
• Snapshot และ Incremental Query
• Indexing

Copy on Write and Merge on Read

ใน Hudi จะ support table type 2 แบบคือ

• Copy on write

เป็น table type หนึ่งที่เก็บข้อมูลในรูปแบบ columnar file format (Parquet file) ซึ่งเป็นแบบที่ง่ายในการ update ข้อมูลและ rewrite ไฟล์เดิมด้วยวิธี synchronous

• Merge on read

เป็น table type อีกแบบที่เก็บทั้งรูปแบบ columnar (Parquet file) และ row based (Avro file) ซึ่งเป็นวิธีที่ complex กว่าแต่ flexible กว่า หากมีการ update ก็จะ log ลงไปใน delta file แล้วค่อยนำมารวมกับ columnar file ทีหลังแบบ synchronous หรือ asynchronous ก็ได้

โดยความน่าสนใจอยู่ตรงที่ Hudi ได้เคลมว่า term นี้ที่ใช้ในฝั่ง Hudi เองถูกใช้ใน Iceberg ด้วย และ ณ เวลาที่เรากำลังเขียนอยู่ ผู้เล่นทั้ง 3 ตัว Hudi, Iceberg และ Delta lake มี Copy On Write ทั้งหมด เพียงแต่ Iceberg และ Delta Lake ยังไม่มี Merge On Read อีกด้วย

Snapshot and Incremental query

Hudi support 3 query types ซึ่งก็คือ

• Snapshot query — การ query เพื่อดู snapshot ของทั้ง table โดยแค่ระบุเลข commit/compaction ที่เราต้องการ

ใน copy-on-write table จะดึง columnar table จากไฟล์ล่าสุด ซึ่งจะ upsert/delete ใน table เดิมได้เลย

ส่วนใน merge-on-read จะทำ snapshot query แบบ near real-time ไม่กี่นาทีด้วยการ merge ตัว base file หลักกับตัว delta file เข้าด้วยกันแบบ on-the-fly

• Incremental query — การ query เพื่อดูเฉพาะข้อมูลใหม่ที่เข้ามาใน table ด้วยการระบุ commit/compaction ที่เราต้องการ
• Read Optimized Queries — คล้ายกับ snapshot query แต่เน้นเรื่องการอ่านจาก parquet มากกว่า

หากผู้อ่านมีความสงสัยว่า columnar หรือ row based file อย่าง Parquet และ Avro คืออะไร สามารถไปศึกษาได้ที่ Big Data file formats: ทำความรู้จัก Avro, Parquet และ ORC

Indexing

จะพูดเรื่อง lakehouse ทั้งที จะไม่พูดถึง Indexing เลยก็ไม่ได้

Hudi ใช้เรื่อง index ในการตามหา file group เพื่อที่จะลด response time และทำให้ upsert/delete เร็วขึ้น

ซึ่งโดย default จะใช้ Bloom filter ในการทำ indexing

เปรียบเทียบการทำ index กับการไม่ใช่ index

merge cost ที่แตกต่างกัน โดยกล่องสีขาวคือ base file, กล่องสีเหลืองคือ updates

ฝั่ง Uber เองก็ใช้เรื่อง indexing เป็นตัวช่วยหลักในการ upsert/delete ข้อมูลในเลเวล large scale ซึ่งสามารถอ่านเพิ่มเติมถึง workload และ experience ละเอียดๆได้ที่ลิ้งก์ด้านล่าง

Employing the right indexes for fast updates, deletes in Apache Hudi | Apache Hudi

ความแตกต่างระหว่าง Hudi, Iceberg, Delta Lake

เราแนะนำให้ลองอ่าน บทความของ Pongthep Vijite ในการเปรียบเทียบ Hudi กับ Iceberg

ความเหมือนที่แตกต่างของจิ๊กซอว์ใน lakehouse อย่าง apache hudi และ apache iceberg

และของ Onehouse.ai เขาเปรียบเทียบฟีเจอร์ต่างๆของทั้ง 3 ตัว

Hudi vs Delta vs Iceberg Lakehouse Feature Comparisons

สรุป

บทความนี้เราได้เล่าถึงการทำ Data Lakehouse โดยเล่าผ่าน Apache Hudi ที่เกิดมาเพื่อ support การทำ streaming ingest เพื่อการ process ข้อมูลแบบ mini-batch และ near real-time โดยเฉพาะ โดยจะเกาะอยู่บน HDFS และ Data Lake เจ้าต่างๆ และเสริมคุณสมบัติด้านการการันตี ACID, จัดการ metadata และ Indexing

รวมถึงเรายังได้รู้ว่าทาง Uber เขาเคยมี pain เรื่องอะไรจนเกิดมาเป็น Apache Hudi ขึ้นมา รวมถึงฟีเจอร์และองค์ประกอบต่างๆใน Hudi ด้วย

Ref.

• https://www.uber.com/en-TH/blog/hoodie/
• https://docs.wso2.com/display/DAS310/Incremental+Processing
• https://www.oreilly.com/content/ubers-case-for-incremental-processing-on-hadoop/
• https://www.linkedin.com/pulse/apache-hudi-streaming-data-lake-platform-vinoth-chandar/
• https://airbyte.com/blog/data-lake-lakehouse-guide-powered-by-table-formats-delta-lake-iceberg-hudi
• https://hudi.apache.org/blog/2021/07/21/streaming-data-lake-platform/
• https://www.uber.com/en-TH/blog/apache-hudi-graduation/
• https://hudi.apache.org/blog/2020/11/11/hudi-indexing-mechanisms/