我正在尝试使用 Airflow SparkSubmitOperatorspark-examples.jar
在本地 Spark 独立上触发,但我不断收到异常。当我在终端上手动提交相同的作业时,它可以工作:
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.
我所做的
$SPARK_HOME/plugins
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,
)
你有一个错字:
java_class='class org.apache.spark.examples.SparkPi'
它应该是 java_class='org.apache.spark.examples.SparkPi'
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句