Airflow using with HDFS and Apache Spark

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.