sirve - settitle java

¿Cómo ejecutar el código Spark en Airflow?

Question

Hola gente de la tierra! Estoy usando Airflow para programar y ejecutar tareas Spark. Todo lo que encontré a estas alturas son los DAG de python que Airflow puede administrar.
Ejemplo DAG:

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
)

El problema es que no soy bueno en el código Python y tengo algunas tareas escritas en Java. Mi pregunta es ¿cómo ejecutar el Spark Java jar en python DAG? O tal vez hay otra manera de hacerlo? Encontré chispa enviar: http://spark.apache.org/docs/latest/submitting-applications.html
Pero no sé cómo conectar todo juntos. Tal vez alguien lo haya usado antes y tenga ejemplo de trabajo. ¡Gracias por tu tiempo!

Answer 1

Deberías poder utilizar BashOperator . Manteniendo el resto de su código tal como está, importe la clase requerida y los paquetes del sistema:

from airflow.operators.bash_operator import BashOperator

import os
import sys

establecer rutas requeridas:

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

y añadir operador:

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
)

Puede extender esto fácilmente para proporcionar argumentos adicionales utilizando plantillas Jinja.

Por supuesto, puede ajustar esto para un escenario que no sea Spark reemplazando bash_command con una plantilla adecuada para su caso, por ejemplo:

bash_command = ''java -jar {{ params.jar }}''

y ajustando params .

Answer 2

El flujo de aire a partir de la versión 1.8 (lanzado hoy), ha

• SparkSqlOperator - https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/spark_sql_operator.py ;

Código SparkSQLHook: https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/hooks/spark_sql_hook.py

• SparkSubmitOperator - https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/spark_submit_operator.py

Código SparkSubmitHook - https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/hooks/spark_submit_hook.py

Tenga en cuenta que estos dos nuevos operadores / ganchos Spark están en la rama "contrib" a partir de la versión 1.8, por lo que no están (bien) documentados.

Así que puedes usar SparkSubmitOperator para enviar tu código java para la ejecución de Spark.

Answer 3

Hay un ejemplo del uso de SparkSubmitOperator para Spark 2.3.1 en kubernetes (instancia de minikube):

Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from airflow.models import Variable
from datetime import datetime, timedelta

default_args = {
    ''owner'': ''[email protected]'',
    ''depends_on_past'': False,
    ''start_date'': datetime(2018, 7, 27),
    ''email'': [''[email protected]''],
    ''email_on_failure'': False,
    ''email_on_retry'': False,
    ''retries'': 1,
    ''retry_delay'': timedelta(minutes=5),
    # ''queue'': ''bash_queue'',
    # ''pool'': ''backfill'',
    # ''priority_weight'': 10,
    ''end_date'': datetime(2018, 7, 29),
}

dag = DAG(
    ''tutorial_spark_operator'', default_args=default_args, schedule_interval=timedelta(1))

t1 = BashOperator(
    task_id=''print_date'',
    bash_command=''date'',
    dag=dag)

print_path_env_task = BashOperator(
    task_id=''print_path_env'',
    bash_command=''echo $PATH'',
    dag=dag)

spark_submit_task = SparkSubmitOperator(
    task_id=''spark_submit_job'',
    conn_id=''spark_default'',
    java_class=''com.ibm.cdopoc.DataLoaderDB2COS'',
    application=''local:///opt/spark/examples/jars/cppmpoc-dl-0.1.jar'',
    total_executor_cores=''1'',
    executor_cores=''1'',
    executor_memory=''2g'',
    num_executors=''2'',
    name=''airflowspark-DataLoaderDB2COS'',
    verbose=True,
    driver_memory=''1g'',
    conf={
        ''spark.DB_URL'': ''jdbc:db2://dashdb-dal13.services.dal.bluemix.net:50001/BLUDB:sslConnection=true;'',
        ''spark.DB_USER'': Variable.get("CEDP_DB2_WoC_User"),
        ''spark.DB_PASSWORD'': Variable.get("CEDP_DB2_WoC_Password"),
        ''spark.DB_DRIVER'': ''com.ibm.db2.jcc.DB2Driver'',
        ''spark.DB_TABLE'': ''MKT_ATBTN.MERGE_STREAM_2000_REST_API'',
        ''spark.COS_API_KEY'': Variable.get("COS_API_KEY"),
        ''spark.COS_SERVICE_ID'': Variable.get("COS_SERVICE_ID"),
        ''spark.COS_ENDPOINT'': ''s3-api.us-geo.objectstorage.softlayer.net'',
        ''spark.COS_BUCKET'': ''data-ingestion-poc'',
        ''spark.COS_OUTPUT_FILENAME'': ''cedp-dummy-table-cos2'',
        ''spark.kubernetes.container.image'': ''ctipka/spark:spark-docker'',
        ''spark.kubernetes.authenticate.driver.serviceAccountName'': ''spark''
        },
    dag=dag,
)

t1.set_upstream(print_path_env_task)
spark_submit_task.set_upstream(t1)

El código que utiliza las variables almacenadas en las variables de flujo de aire:

Además, debe crear una nueva conexión de chispa o editar el ''spark_default'' existente con diccionario adicional {"queue":"root.default", "deploy-mode":"cluster", "spark-home":"", "spark-binary":"spark-submit", "namespace":"default"} :