Apache Airflow 将数据从 BashOperator 传递到 SparkSubmitOperator

Question

我正在尝试登录服务器100.18.10.182，并从Apache Airflow中的.182服务器触发Spark我的Spark提交作业。我已经使用了Bashoperator（Shell脚本将SSH用于100.18.10.182服务器），并且在Spark提交作业中，我使用Sparksubmitoperator作为Bashoperator的下游。 我能够成功地执行Bashoperator，但Spockoperator失败了： 无法执行：spark提交

我认为这是因为我无法通过SSH的会话（of .182服务器）进入下一个Sparksubmitoperator，或者可能是由于与 - JARS或 - 包装有关的其他一些问题，在这里不确定。

我一直在考虑使用XCOM_PUSH将我的bashoperator和XCOM_PULL从SparkSubmitoperator中推入一些数据，但不确定如何以登录服务器的方式传递它，然后我的SparkSubmitoperator从该盒子本身触发？

气流DAG代码：

    t2 = BashOperator(
    task_id='test_bash_operator',
    bash_command="/Users/hardikgoel/Downloads/Work/airflow_dir/shell_files/airflow_prod_ssh_script.sh ",
    dag=dag)
t2

t3_config = {
    'conf': {
        "spark.yarn.maxAppAttempts": "1",
        "spark.yarn.executor.memoryOverhead": "8"
    },
    'conn_id': 'spark_default',
    'packages': 'com.sparkjobs.SparkJobsApplication',
    'jars': '/var/spark/spark-jobs-0.0.1-SNAPSHOT-1/spark-jobs-0.0.1-SNAPSHOT.jar firstJob',
    'driver_memory': '1g',
    'total_executor_cores': '21',
    'executor_cores': 7,
    'executor_memory': '48g'
}


t3 = SparkSubmitOperator(
    task_id='t3',
    **t3_config)

t2 >> t3

外壳脚本代码：

    #!/bin/bash
USERNAME=hardikgoel
HOSTS="100.18.10.182"
SCRIPT="pwd; ls"
ssh -l ${USERNAME} ${HOSTS} "${SCRIPT}"
echo "SSHed successfully"
if [ ${PIPESTATUS[0]} -eq 0 ]; then
  echo "successfull"
fi

Answer 1

将 SSH 和 Spark 提交命令组合在同一个 BashOperator 中：

t2 = BashOperator(
    task_id='ssh_and_spark_submit',
    bash_command="ssh -tt ${USERNAME}@${HOSTS} '/path/to/spark-submit --jars ${JARS} --packages ${PACKAGES} ${SPARK_SUBMIT_ARGS}'",
    dag=dag
)

您还可以使用 xcom：

t2 = BashOperator(
    task_id='ssh_and_push_success',
    bash_command="ssh -tt ${USERNAME}@${HOSTS} 'pwd; ls' && echo 'success'",
    xcom_push=True,
    dag=dag
)

t3 = SparkSubmitOperator(
    task_id='spark_submit_if_ssh_success',
    trigger_rule='one_success',
    provide_context=True,  # Access XCom value
    **t3_config
)

def trigger_spark_if_ssh_success(context):
    return context['ti'].xcom_pull(task_ids='ssh_and_push_success') == 'success'

t3.set_upstream(t2)
t3.set_downstream(TriggerDagRunOperator(trigger_dag_id="downstream_dag_id"))