从 Airflow 作业执行 SparkSubmitOperator 时出错答案

【问题标题】：Error Executing SparkSubmitOperator from Airflow Job从 Airflow 作业执行 SparkSubmitOperator 时出错
【发布时间】：2020-03-18 13:38:20
【问题描述】：

背景：我创建了一个新的 Airflow 作业/任务 DAG，我在其中使用了 SparkSubmitOperator。我在我的桌面上同时运行 Spark 和 Airflow（版本等如下）。 DAG 工作正常，直到它到达 Spark 作业的提交部分。我尝试使用以下选项更改连接。无论我尝试什么，我都会在 Airflow 日志中收到以下消息。

Airflow 识别连接并尝试使用它但失败了。

如果我从命令提示符提交目标 DataPipelineExample.py，它会毫无问题地运行。

问题：是什么阻止 Airflow 识别并使用连接来触发本地火花以执行 spark-submit？

Airflow.exceptions.AirflowException：无法执行：spark-submit --master http://localhost:4040 --name mySparkSubmitJob

桌面：Linux Mint VERSION="19.3 (Tricia)" 火花：版本 2.4.5 Pyspark：版本 2.4.5 气流：版本：1.10.9 Python 3.7.4（默认，2019 年 8 月 13 日，20:35:49） java版本“1.8.0_241”

使用或尝试过的气流连接本地主机 4040 火花：//本地主机4040 http://localhost:4040 http://specificip地址：4040 主机：localhost 端口：4040 / Extras，No Extras 等。附加：{“root.default”，：“spark_home”：“”，“spark_binary”：“spark-submit”，“命名空间”：“default”}

路径信息

export SCALA_HOME=~/anaconda3/share/scala-2.11.1
export SPARK_HOME=/usr/local/spark
export PYSPARK_SUBMIT_ARGS="--master local[2] pyspark-shell"
export PATH=$PATH:/usr/local/spark/bin

低于完整的 DAG。这编译并被 Python 和 Airflow 完全识别。

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': 'me@gmail.com',
    'depends_on_past': False,
    'start_date': datetime(2020, 3, 17),
    'email': ['me@gmail.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    'end_date': datetime(2030, 3, 17),
}

dag = DAG(dag_id = 'a_data_pipelne_job', default_args=default_args, schedule_interval='*/45 * * * *')

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_02',
    conn_id='spark_local',
    application = "/home/me/.config/spyder-py3/DataPipelineExample.py",
    name='airflowspark-DataLoaderMongo',
    verbose=True,
    dag=dag,
)

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

【问题讨论】：

您能分享一下您的spark_local 连接是什么样的吗？如果我没看错，错误消息会说 No Extras 也许你需要定义 Extras

标签： linux apache-spark airflow pyspark-sql

【解决方案1】：

我能够通过使用 SSHOperator 解决这个问题。它比 SparkSubmitOperator 更不容易受到环境配置问题的影响。 SparkSubmit 在本地 pyspark home 的上下文中通过 SSH 调用。为你的 python 脚本添加路径参数，你就可以开始了。

dag = DAG(dag_id = 'a_pjm_data_pipelne__ssh_job', 
                  default_args=default_args, 
                  schedule_interval='*/60 * * * *',
                  params={'project_source': '/home/me/.config/spyder-py3',
                  'spark_submit': '/usr/local/spark/bin/spark-submit DataPipelineExample.py'})

templated_bash_command = """
    echo 'HOSTNAME: localhost' #To check that you are properly connected to the host
    cd {{ params.project_source }}
    {{ params.spark_submit }}
"""

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)

submit_spark_task = SSHOperator(
    task_id="SSH_task",
    ssh_conn_id='ssh_default',
    command=templated_bash_command,
    dag=dag

【讨论】：