MACRO NEPAL
Unlocking Real-Time Data Lakes: Apache Hudi for Upserts in Java
awiskar acharya
November 17, 2025
10:18 am

Traditional data lakes have excelled at handling massive volumes of append-only data but struggled with updates and deletes—operations critical for modern use cases like GDPR compliance, change data capture (CDC), and real-time analytics. Apache Hudi (Hadoop Upserts Deletes and Incrementals) solves this problem by bringing database-like upsert capabilities to data lakes while maintaining scalability and efficiency.

What is Apache Hudi?

Apache Hudi is an open-source data management framework that enables transactional capabilities (ACID compliance) on data lakes. It provides:

  • Upsert Support: Insert new records or update existing ones
  • Incremental Processing: Process only changed data
  • ACID Transactions: Ensure data consistency
  • Schema Evolution: Handle evolving data schemas
  • Multiple Query Engines: Integration with Spark, Flink, Presto, Trino, and Hive

Core Hudi Concepts

  1. Table Types:
  • Copy on Write (CoW): Updates create new file versions (better for read-heavy workloads)
  • Merge on Read (MoR): Updates go to delta logs merged on read (better for write-heavy workloads)
  1. Query Types:
  • Snapshot Queries: Latest committed data
  • Incremental Queries: Only changes since last commit
  • Read Optimized Queries: Latest committed data without merge (MoR only)
  1. Timeline: Hudi maintains a timeline of all table operations for transactional consistency.

Hudi Architecture for Upserts

[Data Source] → [Ingestion Job] → [Hudi Table] → [Query Engine]
|               |                |               |
Kafka          Spark/Flink     Parquet +      Spark/Presto/
JDBC           Java App        .hoodie files  Hive/Trino

Hudi stores data in Parquet files (base data) and Avro log files (updates in MoR), with metadata in .hoodie directory for transaction management.

Hands-On Tutorial: Implementing Hudi Upserts with Java and Spark

Let's build a complete example showing how to perform upserts on a data lake using Hudi, Java, and Apache Spark.

Step 1: Project Setup

Maven Dependencies (pom.xml):

<properties>
<spark.version>3.4.0</spark.version>
<hudi.version>0.13.1</hudi.version>
</properties>
<dependencies>
<!-- Apache Spark -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<!-- Apache Hudi -->
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark3.4-bundle_2.12</artifactId>
<version>${hudi.version}</version>
</dependency>
<!-- AWS SDK for S3 (if using cloud storage) -->
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
<version>2.20.0</version>
</dependency>
</dependencies>

Step 2: Data Model

Let's model a user profile table that needs frequent updates:

public class UserProfile {
private String userId;
private String firstName;
private String lastName;
private String email;
private String lastLogin;
private int loginCount;
private String updatedAt;
// Constructors
public UserProfile() {}
public UserProfile(String userId, String firstName, String lastName, 
String email, String lastLogin, int loginCount, String updatedAt) {
this.userId = userId;
this.firstName = firstName;
this.lastName = lastName;
this.email = email;
this.lastLogin = lastLogin;
this.loginCount = loginCount;
this.updatedAt = updatedAt;
}
// Getters and Setters
public String getUserId() { return userId; }
public void setUserId(String userId) { this.userId = userId; }
public String getFirstName() { return firstName; }
public void setFirstName(String firstName) { this.firstName = firstName; }
public String getLastName() { return lastName; }
public void setLastName(String lastName) { this.lastLogin = lastName; }
public String getEmail() { return email; }
public void setEmail(String email) { this.email = email; }
public String getLastLogin() { return lastLogin; }
public void setLastLogin(String lastLogin) { this.lastLogin = lastLogin; }
public int getLoginCount() { return loginCount; }
public void setLoginCount(int loginCount) { this.loginCount = loginCount; }
public String getUpdatedAt() { return updatedAt; }
public void setUpdatedAt(String updatedAt) { this.updatedAt = updatedAt; }
}

Step 3: Hudi Table Manager

import org.apache.hudi.DataSourceWriteOptions;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import java.util.HashMap;
import java.util.Map;
public class HudiTableManager {
private final SparkSession sparkSession;
private final String basePath;
private final String tableName;
public HudiTableManager(String basePath, String tableName) {
this.basePath = basePath;
this.tableName = tableName;
this.sparkSession = SparkSession.builder()
.appName("HudiUpsertExample")
.master("local[*]")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
.config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.getOrCreate();
}
/**
* Create initial Hudi table with some sample data
*/
public void createInitialTable() {
// Sample initial data
String[][] initialData = {
{"user1", "John", "Doe", "[email protected]", "2024-01-15 10:30:00", "5", "2024-01-15 10:30:00"},
{"user2", "Jane", "Smith", "[email protected]", "2024-01-15 11:45:00", "3", "2024-01-15 11:45:00"},
{"user3", "Bob", "Johnson", "[email protected]", "2024-01-15 09:15:00", "8", "2024-01-15 09:15:00"}
};
// Create DataFrame from sample data
Dataset<Row> df = sparkSession.createDataFrame(
java.util.Arrays.asList(
new UserProfile(initialData[0][0], initialData[0][1], initialData[0][2], 
initialData[0][3], initialData[0][4], Integer.parseInt(initialData[0][5]), initialData[0][6]),
new UserProfile(initialData[1][0], initialData[1][1], initialData[1][2], 
initialData[1][3], initialData[1][4], Integer.parseInt(initialData[1][5]), initialData[1][6]),
new UserProfile(initialData[2][0], initialData[2][1], initialData[2][2], 
initialData[2][3], initialData[2][4], Integer.parseInt(initialData[2][5]), initialData[2][6])
),
UserProfile.class
);
// Write initial data as Hudi table
writeToHudiTable(df, "insert");
System.out.println("Initial Hudi table created successfully!");
}
/**
* Perform upsert operations on Hudi table
*/
public void upsertData(Dataset<Row> upsertDf) {
writeToHudiTable(upsertDf, "upsert");
System.out.println("Upsert operation completed successfully!");
}
/**
* Perform delete operations on Hudi table
*/
public void deleteData(Dataset<Row> deleteDf) {
Map<String, String> options = getBaseHudiOptions();
options.put(DataSourceWriteOptions.OPERATION().key(), "delete");
deleteDf.write()
.format("org.apache.hudi")
.options(options)
.mode(SaveMode.Append)
.save(basePath + "/" + tableName);
System.out.println("Delete operation completed successfully!");
}
/**
* Common method to write data to Hudi table
*/
private void writeToHudiTable(Dataset<Row> df, String operation) {
Map<String, String> options = getBaseHudiOptions();
options.put(DataSourceWriteOptions.OPERATION().key(), operation);
df.write()
.format("org.apache.hudi")
.options(options)
.mode(SaveMode.Append)
.save(basePath + "/" + tableName);
}
/**
* Base Hudi configuration options
*/
private Map<String, String> getBaseHudiOptions() {
Map<String, String> options = new HashMap<>();
// Table configuration
options.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
options.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "userId");
options.put(DataSourceWriteOptions.PRECOMBINE_FIELD().key(), "updatedAt");
options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "");
// Hudi configuration
options.put(HoodieWriteConfig.TBL_NAME.key(), tableName);
options.put(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME().key(), 
"org.apache.hudi.keygen.NonpartitionedKeyGenerator");
// Async services (optional)
options.put("hoodie.cleaner.policy", "KEEP_LATEST_COMMITS");
options.put("hoodie.cleaner.commits.retained", "2");
options.put("hoodie.keep.max.commits", "3");
options.put("hoodie.keep.min.commits", "2");
return options;
}
/**
* Query the Hudi table
*/
public void queryTable() {
Dataset<Row> hudiDf = sparkSession.read()
.format("org.apache.hudi")
.load(basePath + "/" + tableName + "/*");
System.out.println("=== Current Hudi Table Contents ===");
hudiDf.show();
// Show the timeline
Dataset<Row> timelineDf = sparkSession.sql(
"SELECT _hoodie_commit_time, _hoodie_operation FROM hudi_table"
);
System.out.println("=== Hudi Timeline ===");
timelineDf.show();
}
/**
* Read incremental changes
*/
public void readIncrementalChanges(String startTime) {
sparkSession.read()
.format("org.apache.hudi")
.option(DataSourceReadOptions.QUERY_TYPE().key(), 
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL())
.option(DataSourceReadOptions.BEGIN_INSTANTTIME().key(), startTime)
.load(basePath + "/" + tableName)
.show();
}
public void close() {
if (sparkSession != null) {
sparkSession.close();
}
}
}

Step 4: Main Application Class

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
public class HudiUpsertApplication {
private static final DateTimeFormatter formatter = 
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
public static void main(String[] args) {
String basePath = "/tmp/hudi_user_profiles"; // Use S3 path in production: s3a://bucket/path
String tableName = "user_profiles";
HudiTableManager hudiManager = new HudiTableManager(basePath, tableName);
try {
// Step 1: Create initial table
System.out.println("Step 1: Creating initial Hudi table...");
hudiManager.createInitialTable();
hudiManager.queryTable();
// Step 2: Perform upserts - update existing users and add new ones
System.out.println("\nStep 2: Performing upsert operations...");
List<UserProfile> upsertData = Arrays.asList(
// Update existing user
new UserProfile("user1", "John", "Doe-Updated", "[email protected]", 
getCurrentTimestamp(), 6, getCurrentTimestamp()),
// Insert new user
new UserProfile("user4", "Alice", "Brown", "[email protected]", 
getCurrentTimestamp(), 1, getCurrentTimestamp())
);
Dataset<Row> upsertDf = hudiManager.sparkSession.createDataFrame(upsertData, UserProfile.class);
hudiManager.upsertData(upsertDf);
hudiManager.queryTable();
// Step 3: Perform delete operation
System.out.println("\nStep 3: Performing delete operation...");
List<UserProfile> deleteData = Arrays.asList(
new UserProfile("user2", null, null, null, null, 0, getCurrentTimestamp())
);
Dataset<Row> deleteDf = hudiManager.sparkSession.createDataFrame(deleteData, UserProfile.class);
hudiManager.deleteData(deleteDf);
hudiManager.queryTable();
// Step 4: Demonstrate incremental querying
System.out.println("\nStep 4: Reading incremental changes...");
// In real scenario, you would store the commit time from previous operations
hudiManager.readIncrementalChanges("20240101000000000"); // Example start time
} finally {
hudiManager.close();
}
}
private static String getCurrentTimestamp() {
return LocalDateTime.now().format(formatter);
}
}

Step 5: Production Configuration

For production environments, create a configuration class:

public class HudiConfig {
public static Map<String, String> getProductionConfig() {
Map<String, String> config = new HashMap<>();
// AWS S3 Configuration
config.put("hoodie.datasource.write.storage.type", "COPY_ON_WRITE");
config.put("hoodie.datasource.hive_sync.database", "production_db");
config.put("hoodie.datasource.hive_sync.table", "user_profiles");
config.put("hoodie.datasource.hive_sync.enable", "true");
// Performance tuning
config.put("hoodie.parquet.max.file.size", "134217728"); // 128MB
config.put("hoodie.parquet.block.size", "134217728"); // 128MB
config.put("hoodie.parquet.page.size", "1048576"); // 1MB
// Compression
config.put("hoodie.parquet.compression.codec", "gzip");
// Cleaner configuration
config.put("hoodie.cleaner.fileversions.retained", "3");
config.put("hoodie.cleaner.commits.retained", "10");
return config;
}
public static Map<String, String> getAWSS3Config(String bucket, String region) {
Map<String, String> config = new HashMap<>();
config.put("fs.s3a.access.key", "your-access-key");
config.put("fs.s3a.secret.key", "your-secret-key");
config.put("fs.s3a.endpoint", "s3." + region + ".amazonaws.com");
config.put("fs.s3a.region", region);
config.put("hoodie.base.path", "s3a://" + bucket + "/hudi-tables/");
return config;
}
}

Running the Application

  1. Ensure you have Spark installed or run in local mode
  2. Compile and run the application:
   mvn clean compile exec:java -Dexec.mainClass="HudiUpsertApplication"
  1. Observe the output showing:
  • Initial table creation
  • Upsert operations (updates and inserts)
  • Delete operations
  • Table state after each operation

Production Best Practices

1. Table Type Selection

// For read-heavy workloads
options.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
// For write-heavy workloads with frequent updates
options.put(DataSourceWriteOptions.TABLE_TYPE().key(), "MERGE_ON_READ");

2. Partitioning Strategy

// Partition by date for time-series data
options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "createdDate");
options.put(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME().key(), 
"org.apache.hudi.keygen.SimpleKeyGenerator");
// Multi-level partitioning
options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "region,country");

3. Optimized Upserts with Indexing

// Enable Bloom filter indexing for faster upserts
options.put("hoodie.bloom.index.update.partition.path", "true");
options.put("hoodie.bloom.index.bucketized.checking", "true");
options.put("hoodie.bloom.index.prune.by.ranges", "true");

4. Monitoring and Metrics

// Enable metrics collection
options.put("hoodie.metrics.on", "true");
options.put("hoodie.metrics.reporter.type", "GRAPHITE");
options.put("hoodie.metrics.graphite.host", "localhost");
options.put("hoodie.metrics.graphite.port", "2003");

Use Cases for Hudi Upserts

  1. CDC Pipelines: Replicate database changes to data lakes
  2. GDPR Compliance: Handle right-to-be-forgotten requests
  3. Real-time Analytics: Maintain up-to-date dimension tables
  4. Data Deduplication: Remove duplicates during ingestion
  5. Machine Learning: Update feature stores with latest data

Conclusion

Apache Hudi revolutionizes data lake management by bringing database-like upsert capabilities to massive-scale data storage. With Java and Spark integration, you can:

  • Handle real-time updates and deletes efficiently
  • Maintain ACID compliance in data lake environments
  • Enable incremental processing pipelines
  • Scale to petabytes while supporting frequent updates
  • Integrate with modern query engines and cloud platforms

By implementing Hudi with proper configuration and best practices, you can build robust, scalable data lakes that support both batch and real-time use cases while maintaining data quality and consistency.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper