Apache Hudi

Overview

Apache Hudi (Hadoop Upserts Deletes and Incrementals) is an open table format designed for upserting streaming data in real-time.

What You Will Learn

Detailed Concept Explanation

Apache Hudi (Hadoop Upserts Deletes and Incrementals) is a storage layer designed to support stream processing on top of HDFS or cloud storage.

Copy on Write (CoW) vs Merge on Read (MoR)

Hudi supports two table types:

1. Copy on Write (CoW): Every write operation updates files by rewriting them with the new records. This optimizes read performance but is slower for writes.
2. Merge on Read (MoR): Updates are written to separate log files. During reads, Hudi merges these log files with the base parquet files, optimizing write performance at the cost of slower reads.

Code Examples

Input Dataset Preview

Below is the users transaction updates dataset:

Python (PySpark) Implementation

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HudiTest") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .getOrCreate()

data = [(1, 99)]
df = spark.createDataFrame(data, ["userId", "score"])

# Write to Hudi format
df.write.format("hudi") \
    .option("hoodie.datasource.write.recordkey.field", "userId") \
    .option("hoodie.table.name", "hudi_users") \
    .mode("overwrite") \
    .save("hudi_table")

# Read Hudi
hudi_df = spark.read.format("hudi").load("hudi_table")
hudi_df.show()

Expected Output

+------+-----+
|userId|score|
+------+-----+
|     1|   99|
+------+-----+

Execution Plan Diagram (Python & Scala)

Scala Implementation

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("HudiScala")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

import spark.implicits._
val df = Seq((1, 99)).toDF("userId", "score")

df.write.format("hudi")
  .option("hoodie.datasource.write.recordkey.field", "userId")
  .option("hoodie.table.name", "hudi_users")
  .mode("overwrite")
  .save("hudi_table")

val readDF = spark.read.format("hudi").load("hudi_table")
readDF.show()

Expected Output

+------+-----+
|userId|score|
+------+-----+
|     1|   99|
+------+-----+

SQL Implementation

-- Querying Hudi datasets in SQL
SELECT userId, score 
FROM hudi_table;

Expected Output

Executing this SQL query returns:

Common Mistakes

• Forgetting Record Key Option: Attempting to write Hudi tables without defining hoodie.datasource.write.recordkey.field option. Hudi requires record keys to perform upserts, and will fail to write without them.

Best Practices

• Select Table Types Wisely: Use Copy on Write (CoW) tables for read-heavy analytical workloads and Merge on Read (MoR) tables for write-heavy streaming ingestion workloads.

Interview Perspective

Interactive Challenges

Related Topics

• Delta Lake
• Apache Iceberg
• Reading Data