Category: Big data

Article about Big data

Airflow using with HDFS and Apache Spark

Posted on by maximus

Using Apache Airflow with Apache Spark and HDFS typically involves setting up a workflow where Airflow orchestrates the execution of Spark jobs, and the data processed by Spark is stored or retrieved from HDFS. Here’s an overview of how to set up such a system:

Airflow setup

Airflow is used to schedule and orchestrate the workflow. You’ll need to have Apache Airflow installed and configured with the necessary components for your Spark jobs.

Spark setup

Spark needs to be set up in a cluster environment (e.g., standalone mode, YARN, or Kubernetes). The Spark jobs will typically read and write data to/from HDFS.

HDFS setup

HDFS (Hadoop Distributed File System) serves as the distributed storage system where data is stored and retrieved by Spark jobs.

Airflow and Spark Integration

We need a Scala-based Spark application that will read/write data to/from HDFS or perform any transformation required. We will then submit this application using Airflow’s SparkSubmitOperator.

Here’s an example of a simple Scala-based Spark application that reads data from HDFS, processes it, and saves it back to HDFS.

import org.apache.spark.sql.SparkSession

object SparkJob {
  def main(args: Array[String]): Unit = {
    // Initialize Spark session
    val spark = SparkSession.builder
      .appName("Scala Spark Job")
      .config("spark.hadoop.fs.defaultFS", "hdfs://namenode_host:8020")
      .getOrCreate()

    // Read data from HDFS
    val inputData = spark.read.text("hdfs://namenode_host:8020/input_data.txt")

    // Perform transformation (example: convert to uppercase)
    val transformedData = inputData.rdd.map(line => line.getString(0).toUpperCase())

    // Save transformed data back to HDFS
    transformedData.saveAsTextFile("hdfs://namenode_host:8020/output_data")
    
    spark.stop()
  }
}

To build the Scala application we use a build tool like SBT (Scala Build Tool) to package the Scala application as a JAR file.

name := "SparkJob"

version := "1.0"

scalaVersion := "2.12.10"

libraryDependencies += "org.apache.spark" %% "spark-core" % "3.1.2"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.1.2"

Then build the JAR file with the following command:

sbt package

This will create a JAR file under the target/scala-2.12/ directory, e.g., target/scala-2.12/sparkjob_2.12-1.0.jar.

In next step we need to install Apache Airflow and configure the necessary connections (like spark_default connection) to submit the Spark job to our cluster. Once the Scala Spark job is ready and packaged, we can submit it to our Spark cluster using the SparkSubmitOperator in Airflow. Here is how you can do that:

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2025, 6, 2),
}

dag = DAG(
    'scala_spark_hdfs_etl',
    default_args=default_args,
    schedule_interval=None,
)

spark_submit_task = SparkSubmitOperator(
    task_id='submit_spark_job',
    conn_id='spark_default',  # Airflow Spark connection ID
    application='/path/to/target/scala-2.12/sparkjob_2.12-1.0.jar',  # Path to the Scala JAR file
    name='scala-spark-job',
    conf={'spark.hadoop.fs.defaultFS': 'hdfs://namenode_host:8020'},
    dag=dag,
)

spark_submit_task

Explanation:

  • application: Path to the JAR file generated by your Scala application.
  • conn_id: Connection ID for Spark. You may need to configure this in Airflow (this typically points to your Spark cluster, either YARN or standalone).
  • conf: Spark configuration, such as the HDFS URI.
  • task_id: Unique identifier for the Airflow task.

Airflow Workflow

Now that the DAG is set up, Airflow will submit the Spark job (written in Scala) to your Spark cluster. The flow will execute the Spark job as part of the DAG.

  • Scheduling: You can set up a schedule for when to run the Scala-based Spark job.
  • Monitoring: Airflow will monitor the task and show logs, task success/failure, and retries in the UI.

Optional – Handle HDFS Files in Scala:

If you need to interact with HDFS directly within your Scala Spark job (for example, listing files in HDFS or checking for certain files), you can use the HDFS API in Spark. Here’s an example to list HDFS files in your job:

val hdfsPath = "hdfs://namenode_host:8020/input_data/"
val fs = org.apache.hadoop.fs.FileSystem.get(new java.net.URI(hdfsPath), spark.sparkContext.hadoopConfiguration)
val status = fs.listStatus(new org.apache.hadoop.fs.Path(hdfsPath))

status.foreach(file => println(file.getPath))

Summary:

  1. Scala Application: Write your Spark job in Scala and package it as a JAR.
  2. Airflow DAG: Use the SparkSubmitOperator to submit your Scala-based Spark job.
  3. HDFS: Use HDFS paths and configurations in your Spark job to read/write data.
  4. Scheduling and Monitoring: Airflow handles scheduling and monitoring the execution of your Spark jobs.

Apache Iceberg and time travel queries with Scala/Spark

Posted on by maximus

Iceberg provides time travel functionality, allowing you to query the state of a table at a previous point in time. You can query the table as it existed at a specific snapshot or as of a certain timestamp.

Example: querying an Iceberg table at a specific snapshot

Each time you make changes (like insert or update) to an Iceberg table, a snapshot is created. You can query data from a particular snapshot.

def timeTravelBySnapshot(spark: SparkSession, snapshotId: Long): Unit = {
  val timeTravelDF = spark.read
    .option("snapshot-id", snapshotId)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved for snapshot: $snapshotId")
}

We can find the snapshot ID using:

val historyDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_table.history")
historyDF.show()

Once we have the snapshot ID, pass it to timeTravelBySnapshot() to query the data as of that snapshot.

Example: querying an Iceberg table as of a timestamp

def timeTravelByTimestamp(spark: SparkSession, timestamp: String): Unit = {
  val timeTravelDF = spark.read
    .option("as-of-timestamp", timestamp)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved as of timestamp: $timestamp")
}
// Example timestamp format: "2024-10-16T15:00:00.000Z"
// To execute:
timeTravelByTimestamp(spark, "2024-10-16T15:00:00.000Z")

In this way we can to handle time travel queries using Apache Iceberg and Spark.

Apache Iceberg partitioning data with Scala/Spark

Posted on by maximus

Introduction

Partitioning helps improve query performance by reducing the amount of data scanned during read operations. Iceberg allows dynamic partitioning and supports various partition types such as identity, bucket, and truncate.

Example: partitioning data by a column (e.g., Age)

Let’s partition the table by a column like age (bucket partitioning).

def partitionTable(spark: SparkSession): Unit = {
  // Create a new partitioned table
  spark.sql(
    """
      |CREATE TABLE IF NOT EXISTS spark_catalog.default.iceberg_partitioned_table (
      | id BIGINT,
      | name STRING,
      | age INT
      |) USING iceberg
      | PARTITIONED BY (bucket(5, age))
      |""".stripMargin)
  println("Partitioned Iceberg table created!")
}
// To execute:
partitionTable(spark)

Here, we’ve partitioned the table using the age column into 5 buckets. This is useful for optimizing queries when filtering based on age.

Query example with partitioned table:

When you query partitioned data, Iceberg only scans the partitions relevant to the query:

def queryPartitionedTable(spark: SparkSession): Unit = {
  val partitionedDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_partitioned_table WHERE age > 25")
  partitionedDF.show()
}
// To execute:
queryPartitionedTable(spark)

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.