How to run Spark Java code in Airflow?
<p>Hello people of the Earth!
I'm using Airflow to schedule and run Spark tasks.
All I found by this time is python DAGs that Airflow can manage.
<br> DAG example:</p>

<pre><code>spark_count_lines.py
import logging

from airflow import DAG
from airflow.operators import PythonOperator

from datetime import datetime

args = {
  'owner': 'airflow'
  , 'start_date': datetime(2016, 4, 17)
  , 'provide_context': True
}

dag = DAG(
  'spark_count_lines'
  , start_date = datetime(2016, 4, 17)
  , schedule_interval = '@hourly'
  , default_args = args
)

def run_spark(**kwargs):
  import pyspark
  sc = pyspark.SparkContext()
  df = sc.textFile('file:///opt/spark/current/examples/src/main/resources/people.txt')
  logging.info('Number of lines in people.txt = {0}'.format(df.count()))
  sc.stop()

t_main = PythonOperator(
  task_id = 'call_spark'
  , dag = dag
  , python_callable = run_spark
)
</code></pre>

<p>The problem is I'm not good in Python code and have some tasks written in Java. My question is how to run Spark Java jar in python DAG? Or maybe there is other way yo do it? I found spark submit: <a href="http://spark.apache.org/docs/latest/submitting-applications.html" rel="nofollow">http://spark.apache.org/docs/latest/submitting-applications.html</a>
<br>But I don't know how to connect everything together. Maybe someone used it before and has working example. Thank you for your time!</p>

<p>You should be able to use <code>BashOperator</code>. Keeping the rest of your code as is, import required class and system packages:</p>

<pre><code>from airflow.operators.bash_operator import BashOperator

import os
import sys
</code></pre>

<p>set required paths:</p>

<pre><code>os.environ['SPARK_HOME'] = '/path/to/spark/root'
sys.path.append(os.path.join(os.environ['SPARK_HOME'], 'bin'))
</code></pre>

<p>and add operator:</p>

<pre><code>spark_task = BashOperator(
    task_id='spark_java',
    bash_command='spark-submit --class {{ params.class }} {{ params.jar }} ',
    params={'class': 'MainClassName', 'jar': '/path/to/your.jar'},
    dag=dag
)
</code></pre>

<p>You can easily extend this to provide additional arguments using Jinja templates.</p>

java python apache-spark directed-acyclic-graphs airflow