通过 Airflow 调度在 Kubernetes 上运行的 Spark 作业

Question

我有一个通过 Kubernetes pod 运行的 spark 作业。到目前为止，我一直在使用 Yaml 文件手动运行我的作业。现在，我想通过气流安排我的火花作业。这是我第一次使用气流，我无法弄清楚如何在气流中添加我的 Yaml 文件。据我所知，我可以通过 Airflow 中的 DAG 安排我的工作。一个 dag 示例是这样的：

from airflow.operators import PythonOperator
from airflow.models import DAG
from datetime import datetime, timedelta

args = {'owner':'test', 'start_date' : datetime(2019, 4, 3), 'retries': 2, 'retry_delay': timedelta(minutes=1) }
dag = DAG('test_dag', default_args = args, catchup=False)

def print_text1():
    print("hell-world1")

def print_text():
    print('Hello-World2')

t1 = PythonOperator(task_id='multitask1', python_callable=print_text1, dag=dag)
t2 = PythonOperator(task_id='multitask2', python_callable=print_text, dag=dag)
t1 >> t2

在这种情况下，一旦我播放 DAG，上述方法将依次执行。现在，如果我想运行 spark 提交作业，我该怎么办？我正在使用 Spark 2.4.4

Answer 1

Airflow 有一个operators的概念，它代表 Airflow 任务。在您的示例中使用了PythonOperator，它只是执行 Python 代码，并且很可能不是您感兴趣的代码，除非您在 Python 代码中提交 Spark 作业。您可以使用多种运算符：

• BashOperator，它为您执行给定的 bash 脚本。你可以运行kubectl或spark-submit直接使用它
• SparkSubmitOperator，要调用的特定操作符spark-submit
• KubernetesPodOperator，为您创建Kubernetes pod，您可以直接使用它启动您的Driver pod
• 混合解决方案，例如。HttpOperator + Livy on Kubernetes，您可以在 Kubernetes 上启动 Livy 服务器，该服务器充当 Spark 作业服务器并提供可供 Airflow HttpOperator 调用的 REST API

注意：对于每个操作符，您需要确保您的 Airflow 环境包含执行所需的所有依赖项以及配置为访问所需服务的凭据。

您也可以参考现有线程：

• Airflow SparkSubmitOperator - 如何在另一台服务器中触发提交