由于 java.lang.ClassNotFoundException 导致 Airflow SparkSubmitOperator 失败：类 org.apache.spark.examples.SparkPi

Question

我正在尝试使用 Airflow SparkSubmitOperator 在本地 Spark 独立设备上触发 spark-examples.jar，但我不断收到异常。当我在终端上手动提交相同的作业时，它可以工作：

spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://<HOSTNAME>:7077 \
  --deploy-mode cluster \
  --executor-memory 1G \
  --total-executor-cores 1 \
  /path/to/spark-examples_2.11-2.3.1.jar \
  1000

我想我在气流方面做得不对，但我还没有弄清楚。这是堆栈跟踪：

airflow.exceptions.AirflowException: Cannot execute: ['spark-submit', '--master', 'local', '--conf', 'master=spark://<HOSTNAME>:7077', '--num-executors', '1', '--total-executor-cores', '1', '--executor-cores', '1', '--executor-memory', '2g', '--driver-memory', '1g', '--name', u'airflow-spark-example', '--class', 'class org.apache.spark.examples.SparkPi', '--queue', u'root.default', 'path/to/spark-examples_2.11-2.3.1.jar', u'1000']. Error code is: 101.

我做了什么

1. 我将spark_submit_operator.py 复制到$SPARK_HOME/plugins
2. 我编辑了 spark_default 连接：
   • host: local
   • Extra: {"queue": "root.default", "deploy_mode": "cluster", "spark_home": "", "spark_binary": "spark-submit", "namespace": "default"}

我的朋友：

from airflow import DAG

from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta


args = {
    'owner': 'airflow',
    'start_date': datetime(2018, 7, 31)
}
dag = DAG('spark_example_new', default_args=args, schedule_interval="*/10 * * * *")

operator = SparkSubmitOperator(
    task_id='spark_submit_job',
    conn_id='spark_default',
    java_class='class org.apache.spark.examples.SparkPi',
    application='/path/to/spark-examples_2.11-2.3.1.jar',
    total_executor_cores='1',
    executor_cores='1',
    executor_memory='2g',
    num_executors='1',
    name='airflow-spark-example',
    verbose=False,
    driver_memory='1g',
    application_args=["1000"],
    conf={'master':'spark://<HOSTNAME>:7077'},
    dag=dag,
)

Answer 1

你有一个错字：

java_class='class org.apache.spark.examples.SparkPi'

应该是java_class='org.apache.spark.examples.SparkPi'