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:<\/p>\n\n\n\n
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.<\/p>\n\n\n\n
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.<\/p>\n\n\n\n
HDFS (Hadoop Distributed File System) serves as the distributed storage system where data is stored and retrieved by Spark jobs.<\/p>\n\n\n\n
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 Here\u2019s an example of a simple Scala-based Spark application that reads data from HDFS, processes it, and saves it back to HDFS.<\/p>\n\n\n\n To build the Scala application we use a build tool like SBT<\/strong> (Scala Build Tool) to package the Scala application as a JAR file.<\/p>\n\n\n\n Then build the JAR file with the following command:<\/p>\n\n\n\n This will create a JAR file under the In next step we need to install Apache Airflow and configure the necessary connections (like Airflow Workflow<\/strong><\/p>\n\n\n\n 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.<\/p>\n\n\n\n Optional – Handle HDFS Files in Scala<\/strong>:<\/p>\n\n\n\n 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:<\/p>\n\n\n\n <\/p>\n","protected":false},"excerpt":{"rendered":" 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 … Continue reading Airflow using with HDFS and Apache Spark<\/span> SparkSubmitOperator<\/code>.<\/p>\n\n\n\nimport org.apache.spark.sql.SparkSession\n\nobject SparkJob {\n def main(args: Array[String]): Unit = {\n \/\/ Initialize Spark session\n val spark = SparkSession.builder\n .appName(\"Scala Spark Job\")\n .config(\"spark.hadoop.fs.defaultFS\", \"hdfs:\/\/namenode_host:8020\")\n .getOrCreate()\n\n \/\/ Read data from HDFS\n val inputData = spark.read.text(\"hdfs:\/\/namenode_host:8020\/input_data.txt\")\n\n \/\/ Perform transformation (example: convert to uppercase)\n val transformedData = inputData.rdd.map(line => line.getString(0).toUpperCase())\n\n \/\/ Save transformed data back to HDFS\n transformedData.saveAsTextFile(\"hdfs:\/\/namenode_host:8020\/output_data\")\n \n spark.stop()\n }\n}<\/pre>\n\n\n\nname := \"SparkJob\"\n\nversion := \"1.0\"\n\nscalaVersion := \"2.12.10\"\n\nlibraryDependencies += \"org.apache.spark\" %% \"spark-core\" % \"3.1.2\"\nlibraryDependencies += \"org.apache.spark\" %% \"spark-sql\" % \"3.1.2\"<\/pre>\n\n\n\n
sbt package<\/pre>\n\n\n\n
target\/scala-2.12\/<\/code> directory, e.g., target\/scala-2.12\/sparkjob_2.12-1.0.jar<\/code>.<\/p>\n\n\n\nspark_default<\/code> 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<\/code> in Airflow. Here is how you can do that:<\/p>\n\n\n\nfrom airflow import DAG\nfrom airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator\nfrom datetime import datetime\n\ndefault_args = {\n 'owner': 'airflow',\n 'start_date': datetime(2025, 6, 2),\n}\n\ndag = DAG(\n 'scala_spark_hdfs_etl',\n default_args=default_args,\n schedule_interval=None,\n)\n\nspark_submit_task = SparkSubmitOperator(\n task_id='submit_spark_job',\n conn_id='spark_default', # Airflow Spark connection ID\n application='\/path\/to\/target\/scala-2.12\/sparkjob_2.12-1.0.jar', # Path to the Scala JAR file\n name='scala-spark-job',\n conf={'spark.hadoop.fs.defaultFS': 'hdfs:\/\/namenode_host:8020'},\n dag=dag,\n)\n\nspark_submit_task<\/pre>\n\n\n\nExplanation:<\/h3>\n\n\n\n
\n
application<\/code><\/strong>: Path to the JAR file generated by your Scala application.<\/li>\n\n\n\nconn_id<\/code><\/strong>: Connection ID for Spark. You may need to configure this in Airflow (this typically points to your Spark cluster, either YARN or standalone).<\/li>\n\n\n\nconf<\/code><\/strong>: Spark configuration, such as the HDFS URI.<\/li>\n\n\n\ntask_id<\/code><\/strong>: Unique identifier for the Airflow task.<\/li>\n<\/ul>\n\n\n\n\n
val hdfsPath = \"hdfs:\/\/namenode_host:8020\/input_data\/\"\nval fs = org.apache.hadoop.fs.FileSystem.get(new java.net.URI(hdfsPath), spark.sparkContext.hadoopConfiguration)\nval status = fs.listStatus(new org.apache.hadoop.fs.Path(hdfsPath))\n\nstatus.foreach(file => println(file.getPath))<\/pre>\n\n\n\n
Summary:<\/h3>\n\n\n\n
\n
SparkSubmitOperator<\/code> to submit your Scala-based Spark job.<\/li>\n\n\n\n