Month: August 2024

Apache Hudi CDC (Change Data Capture) Use case with Scala/Spark

Posted on by maximus

Apache Hudi (Hadoop Upserts and Incrementals) integrates with Apache Spark, making it possible to perform Change Data Capture (CDC) using Spark and Scala. Hudi enables incremental data processing by tracking changes (inserts, updates, deletes) and efficiently managing these on top of distributed storage like HDFS or cloud stores (e.g., S3, GCS).

In many business scenarios, we need to capture changes (inserts, updates, and deletes) from a source system and apply them incrementally to your data lake. Hudi’s upsert and incremental queries are well-suited for this scenario.

Set up Apache Hudi and Spark dependencies:

libraryDependencies ++= Seq(
  "org.apache.hudi" %% "hudi-spark3-bundle" % "X.Y.Z",
  "org.apache.spark" %% "spark-core" % "3.3.0",
  "org.apache.spark" %% "spark-sql" % "3.3.0"
)

Replace X.Y.Z with the appropriate version of Hudi and ensure Spark is aligned with the same version.

Create Hudi Table in Spark:

To manage updates and inserts in a table (i.e., CDC), Hudi uses two main table types:

  • Copy on Write (COW): More write-intensive, but gives consistent reads.
  • Merge on Read (MOR): Faster writes but slightly complex reads due to incremental merges.

You can create a Hudi table using Spark DataFrame API. First, load your data into a DataFrame, and then apply Hudi configurations.

import org.apache.spark.sql.{SaveMode, SparkSession}
val spark = SparkSession.builder
  .appName("Hudi CDC Example")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()
// Example data in DataFrame (e.g., CDC data)
val df = spark.read.json("path_to_cdc_data")
// Hudi table write options
val hudiOptions = Map[String, String](
  "hoodie.table.name" -> "hudi_cdc_table",
  "hoodie.datasource.write.operation" -> "upsert",  // To handle CDC events
  "hoodie.datasource.write.recordkey.field" -> "primary_key_column",  // Primary key for CDC
  "hoodie.datasource.write.precombine.field" -> "timestamp_column",   // Helps resolve duplicate rows
  "hoodie.datasource.hive_sync.enable" -> "true",                     // Optional if syncing to Hive
  "hoodie.datasource.hive_sync.database" -> "hudi_db",
  "hoodie.datasource.hive_sync.table" -> "hudi_cdc_table",
  "hoodie.datasource.hive_sync.partition_fields" -> "partition_column",
  "hoodie.datasource.write.hive_style_partitioning" -> "true"
)
// Writing to Hudi table using DataFrame API
df.write.format("hudi")
  .options(hudiOptions)
  .mode(SaveMode.Append)  // Append or Overwrite as needed
  .save("path_to_hudi_table")

Read Incremental Data

After loading data incrementally using Hudi, you can read the changes (new inserts, updates, and deletes) since the last commit.

// Reading Hudi data incrementally from the latest commit
val incrementalOptions = Map(
  "hoodie.datasource.query.type" -> "incremental",
  "hoodie.datasource.read.begin.instanttime" -> "20230101000000" // Specify the commit timestamp to start from
)
val incrementalDF = spark.read.format("hudi")
  .options(incrementalOptions)
  .load("path_to_hudi_table")
incrementalDF.show(false)

Handle Updates and Deletes in CDC

Apache Hudi allows you to manage updates and deletes through the upsert operation. When upserting data, Hudi automatically checks for the primary key and handles whether it should insert, update, or delete the record.

Handling Deletes

For deletions, the key field should be provided, and the operation should be set to delete in Hudi write options.

val deleteOptions = Map[String, String](
  "hoodie.table.name" -> "hudi_cdc_table",
  "hoodie.datasource.write.operation" -> "delete",  // For deletes
  "hoodie.datasource.write.recordkey.field" -> "primary_key_column"
)
val deleteDF = // Your dataframe containing the primary keys to delete
deleteDF.write.format("hudi")
  .options(deleteOptions)
  .mode(SaveMode.Append)
  .save("path_to_hudi_table")

Key Configurations

  • hoodie.datasource.write.operation: Choose upsert, insert, or delete depending on your CDC operation,
  • hoodie.datasource.write.recordkey.field: Set this to the primary key column, which is used to identify unique rows,
  • hoodie.datasource.write.precombine.field: Used for deduplication in the case of multiple updates to the same record,
  • hoodie.datasource.query.type: Used when reading, with options like snapshot (full read) or incremental (read only changed records).

Advantages of Using Apache Hudi for CDC

  1. Efficient Storage: Hudi manages the data lifecycle by compacting files and removing unnecessary duplicates.
  2. Streaming Use Case: With Hudi, you can efficiently consume CDC streams (e.g., from Kafka) and apply updates.
  3. Incremental Processing: Hudi allows for reading incremental changes, which improves processing efficiency when only new or changed data needs to be processed.

Example with Spark Structured Streaming (CDC Streaming)

If you’re processing CDC events from a stream (like Kafka), you can combine Spark Structured Streaming with Hudi as follows:

val streamDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "cdc-topic")
  .load()
// Write stream to Hudi table
streamDF.writeStream
  .format("hudi")
  .options(hudiOptions)
  .outputMode("append")
  .start("path_to_hudi_table")

This allows for continuous ingestion of changes and real-time updates into your Hudi dataset.

Summary

By combining Apache Hudi with Spark in Scala, you can efficiently implement a CDC system that supports upserts, deletes, and real-time data ingestion with minimal overhead. Hudi simplifies the management of large-scale distributed datasets by offering incremental data processing and optimized storage.

The Circuit breaker pattern in Java

Posted on by maximus

The Circuit Breaker pattern is used to prevent an application from performing an operation that is likely to fail. In microservices architecture, it’s commonly used to handle fault tolerance when calling external services that may fail or take a long time to respond.

Imagine a scenario where a microservice (Service A) calls another service (Service B). If Service B is down or slow to respond, without a circuit breaker, Service A will continue making requests, which may lead to cascading failures or resource exhaustion. The Circuit Breaker pattern “opens” and stops calling the failing service for a period to allow it to recover, then “closes” after the service is healthy again.

Key states of Circuit Breaker:

  1. Closed: The circuit breaker allows requests to pass through as normal.
  2. Open: The circuit breaker prevents requests from being sent, returning a fallback response immediately.
  3. Half-Open: The circuit breaker allows a limited number of requests to check if the service has recovered.

Implementation

There are different ways to implement Circuit Breaker pattern. In this case we’ll use Resilience4j for implementing the Circuit Breaker in Spring Boot. Add the following dependency to your pom.xml:

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>1.7.0</version>
</dependency>

Here’s an example of how we might use a RestTemplate to call another service:

import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import org.springframework.beans.factory.annotation.Autowired;
@Service
public class ExternalService {
    @Autowired
    private RestTemplate restTemplate;
    public String callExternalService() {
        String url = "http://localhost:8081/external-api";
        return restTemplate.getForObject(url, String.class);
    }
}

This method will call an external API, but what happens if this API is down or slow? This is where the Circuit Breaker comes in.

To apply the Circuit Breaker, annotate the method that makes the external call using the @CircuitBreaker annotation.

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import org.springframework.stereotype.Service;
@Service
public class ExternalService {
    @Autowired
    private RestTemplate restTemplate;
    private static final String EXTERNAL_SERVICE = "externalService";
    @CircuitBreaker(name = EXTERNAL_SERVICE, fallbackMethod = "fallbackResponse")
    public String callExternalService() {
        String url = "http://localhost:8081/external-api";
        return restTemplate.getForObject(url, String.class);
    }
    public String fallbackResponse(Throwable t) {
        return "External Service is currently unavailable. Please try again later.";
    }
}

@CircuitBreaker(name = "externalService"): This specifies that Resilience4j will manage the circuit breaker with the name externalService.

fallbackMethod = "fallbackResponse": If the circuit breaker is in the OPEN state, or if the call to the external service fails, this fallback method will be executed.

Now we can configure the circuit breaker properties in our application.yml file.

resilience4j.circuitbreaker:
  instances:
    externalService:
      slidingWindowSize: 5
      failureRateThreshold: 50
      waitDurationInOpenState: 10000
      permittedNumberOfCallsInHalfOpenState: 3
  • slidingWindowSize: The number of calls to evaluate for the failure rate,
  • failureRateThreshold: The percentage of failures that trip the circuit breaker. In this case, if 50% of calls fail, the circuit opens,
  • waitDurationInOpenState: The amount of time (in milliseconds) the circuit breaker stays open before transitioning to half-open state,
  • permittedNumberOfCallsInHalfOpenState: The number of calls to allow in the half-open state to check if the external service has recovered.

How it works

  1. Initially, the circuit breaker is in the Closed state. All requests pass through to the external service.
  2. If the failure rate (e.g., timeouts or errors) exceeds the failureRateThreshold, the circuit breaker transitions to the Open state. In this state, all requests are blocked, and the fallback method is executed.
  3. After the waitDurationInOpenState, the circuit breaker transitions to the Half-Open state, allowing a limited number of requests to check if the service has recovered.
  4. If the service responds successfully during the Half-Open state, the circuit breaker transitions back to Closed. Otherwise, it reopens.

Summary

The Circuit Breaker pattern prevents continuous calls to a failing service, providing a graceful degradation (using the fallback method) while the service recovers. Using Resilience4j with Spring Boot simplifies the implementation, allowing you to control fault tolerance with minimal effort.