如何使用 KubernetesPodOperator 编写 Airflow DAG 提交 Spark Jobs

Question

我偶然发现了很多关于如何使用 KubernetesPodOperator 编写 Airflow DAG 来提交 Spark 作业的答案。在挣扎了几天并发现语法不正确之后，不知何故我让它工作了。在这里发布一个答案，这样其他人就不会太挣扎了:)

Answer 1

Below Airflow DAG 使用 KubernetesPodOperator 提交 Spark 作业，其中它从 Ceph 读取 PySpark 脚本。如果您的 PySpark 脚本位于 AWS S3 中，则应该使用相同的语法。对于 AWS S3，您无需传递 fs.s3a.endpoint、fs.s3a.connection.*、f3.s3a.path.*

# Airflow DEMO DAG

from airflow import DAG
from datetime import timedelta, datetime
from kubernetes.client import models as k8s
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator

args = {
    "owner": "prateek.dubey",
    "email": ["<your_email_id>"],
    "depends_on_past": False,
    "start_date": datetime(2019,1,1),
    "catchup": False, 
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5)
}

dag = DAG(dag_id='kubernetes_sample_dag', default_args=args, schedule_interval=None)

ceph_data_read = KubernetesPodOperator(
        namespace='airflow',
        image='spark-executor-3.0.1',
        image_pull_policy='Always',
        image_pull_secrets=[k8s.V1LocalObjectReference('gcr')],
        service_account_name='spark',
        name='prateek-ceph-data-read',
        task_id='ceph_data_read',
        in_cluster=True,
        get_logs=True,
        arguments=[
                '/opt/spark/bin/spark-submit',
                '--master', 'k8s://https://<api_server_host>:6443',
                '--deploy-mode', 'cluster',
                '--name', 'prateek-ceph-data-read',
                '--conf', 'spark.kubernetes.driver.pod.name=prateek-ceph-data-read',
                '--conf', 'spark.kubernetes.executor.podNamePrefix=prateek-ceph-data-read',
                '--conf', 'spark.kubernetes.namespace=airflow',
                '--conf', 'spark.kubernetes.container.image=spark-executor-3.0.1',
                '--conf', 'spark.kubernetes.container.image.pullPolicy=Always',
                '--conf', 'spark.kubernetes.container.image.pullSecrets=gcr',
                '--conf', 'spark.kubernetes.authenticate.driver.serviceAccountName=spark',
                '--conf', 'spark.kubernetes.authenticate.executor.serviceAccountName=spark',
                '--conf', 'spark.kubernetes.authenticate.submission.caCertFile=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt',
                '--conf', 'spark.kubernetes.authenticate.submission.oauthTokenFile=/var/run/secrets/kubernetes.io/serviceaccount/token',
                '--conf', 'spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.EnvironmentVariableCredentialsProvider',
                '--conf', 'spark.hadoop.fs.s3a.endpoint=http://<ceph_endpoint_url>:8080',
                '--conf', 'spark.hadoop.fs.s3a.connection.ssl.enabled=false',
                '--conf', 'spark.hadoop.fs.s3a.path.style.access=true',
                '--conf', 'spark.executor.instances=2',
                '--conf', 'spark.executor.cores=6',
                '--conf', 'spark.executor.memory=55g',
                's3a://<ceph_bucket>/Ceph_PySpark_Read_Data.py'
            ],
        dag=dag
    )

ceph_data_read