Airflow SparkSubmitOperator-如何在另一台服务器中进行火花提交

Question

我是Airflow和Spark的新手，并且正与SparkSubmitOperator苦苦挣扎。

我们的气流调度程序和我们的hadoop集群不在同一台机器上设置（第一个问题：这是一个好习惯吗？）。

我们有许多自动过程需要调用pyspark脚本。这些pyspark脚本存储在hadoop集群（10.70.1.35）中。气流开关存储在气流机（10.70.1.22）中。

当前，当我们想通过气流火花提交pyspark脚本时，我们使用一个简单的BashOperator，如下所示：

cmd = "ssh hadoop@10.70.1.35 spark-submit \
   --master yarn \
   --deploy-mode cluster \
   --executor-memory 2g \
   --executor-cores 2 \
   /home/hadoop/pyspark_script/script.py"
t = BashOperator(task_id='Spark_datamodel',bash_command=cmd,dag=dag)

它工作得很好。但是我们想开始使用SparkSubmitOperator来提交pyspark脚本。

我尝试了这个：

from airflow import DAG
from datetime import timedelta, datetime
from airflow.contrib.operators.spark_submit_operator import 
SparkSubmitOperator
from airflow.operators.bash_operator import BashOperator
from airflow.models import Variable

dag = DAG('SPARK_SUBMIT_TEST',start_date=datetime(2018,12,10), 
schedule_interval='@daily')


sleep = BashOperator(task_id='sleep', bash_command='sleep 10',dag=dag)

_config ={'application':'hadoop@10.70.1.35:/home/hadoop/pyspark_script/test_spark_submit.py',
    'master' : 'yarn',
    'deploy-mode' : 'cluster',
    'executor_cores': 1,
    'EXECUTORS_MEM': '2G'
}

spark_submit_operator = SparkSubmitOperator(
    task_id='spark_submit_job',
    dag=dag,
    **_config)

sleep.set_downstream(spark_submit_operator) 

语法应该可以，因为dag不会显示为已损坏。但是运行时会出现以下错误：

[2018-12-14 03:26:42,600] {logging_mixin.py:95} INFO - [2018-12-14 
03:26:42,600] {base_hook.py:83} INFO - Using connection to: yarn
[2018-12-14 03:26:42,974] {logging_mixin.py:95} INFO - [2018-12-14 
03:26:42,973] {spark_submit_hook.py:283} INFO - Spark-Submit cmd: 
['spark-submit', '--master', 'yarn', '--executor-cores', '1', '--name', 
'airflow-spark', '--queue', 'root.default', 
'hadoop@10.70.1.35:/home/hadoop/pyspark_script/test_spark_submit.py']
[2018-12-14 03:26:42,977] {models.py:1760} ERROR - [Errno 2] No such 
file or directory: 'spark-submit'
Traceback (most recent call last):
      File "/home/dataetl/anaconda3/lib/python3.6/site- 
   packages/airflow/models.py", line 1659, in _run_raw_task    
    result = task_copy.execute(context=context)
      File "/home/dataetl/anaconda3/lib/python3.6/site- 
   packages/airflow/contrib/operators/spark_submit_operator.py", line 
168, 
    in execute
        self._hook.submit(self._application)
      File "/home/dataetl/anaconda3/lib/python3.6/site- 
   packages/airflow/contrib/hooks/spark_submit_hook.py", line 330, in 
submit
        **kwargs)
      File "/home/dataetl/anaconda3/lib/python3.6/subprocess.py", line 
707, 
    in __init__
        restore_signals, start_new_session)
      File "/home/dataetl/anaconda3/lib/python3.6/subprocess.py", line 
    1326, in _execute_child
        raise child_exception_type(errno_num, err_msg)
    FileNotFoundError: [Errno 2] No such file or directory: 'spark-submit'

这是我的问题：

1. 我应该在气流机器上安装spark hadoop吗？我之所以问是因为，在这个主题中，我读到我需要复制hdfs-site.xml和hive-site.xml。但是您可以想象，气流机器上既/etc/hadoop/没有/etc/hive/目录，也没有目录。

2. a）如果否，我应该在哪里复制hdfs-site.xml并hive-site.xml在气流机器上复制？

3. b）如果是，这是否意味着我需要将气流机配置为客户端？一种不参与作业但可以用来提交动作的边缘节点？

4. 那我可以spark-submit从气流机上取下来吗？如果是，那么我不需要像我对mysql数据库那样在Airflow上创建连接，对吗？

5. 哦，蛋糕上的樱桃：我可以将pyspark脚本存储在我的气流机中，spark-submit并从同一台气流机中存储它们。这将是惊人的！

即使您无法回答我所有的问题，任何评论都将非常有用...

还是要提前谢谢！:)

Answer 1

要回答您的第一个问题，是的，这是一个好习惯。

有关如何使用的信息SparkSubmitOperator，请参阅/sf/answers/3734129941/上的我的答案

1. 是的，您需要气流机器上的火花二进制文件。
2. --
3. 是
4. 否 ->您仍然需要一个连接来告诉Airflow您在哪里安装了火花二进制文件。类似于/sf/answers/3537914831/
5. 应该管用