通过 Airflow SparkKubernetesOperator 传递参数-Python教程-PHP中文网

通过 Airflow SparkKubernetesOperator 传递参数

心靈之曲

发布： 2025-08-13 17:18:02

原创

308人浏览过

通过 airflow sparkkubernetesoperator 传递参数

在使用 Airflow 的 SparkKubernetesOperator 时，正确地将参数传递给 Spark 应用程序至关重要。以下将详细介绍如何配置 Airflow DAG 和 SparkApplication YAML 文件，以及如何在 Spark 应用程序中读取这些参数。

配置 Airflow DAG

首先，需要在 Airflow DAG 中定义 SparkKubernetesOperator 任务，并使用 params 参数传递参数。params 应该是一个字典，其中键是参数名称，值是参数值。

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

with DAG('spark_application', default_args=default_args, schedule_interval=None, catchup=False) as dag:

    arguments_to_pass = {
        'id': '1'
    }

    spark_operator = KubernetesPodOperator(  # Changed to KubernetesPodOperator as SparkKubernetesOperator is not standard
        task_id='spark_submit_task',
        namespace='your_namespace',  # 更新为你的 Kubernetes 命名空间
        name="spark-pi",
        image="your_spark_image",
        cmds=["/opt/spark/bin/spark-submit"],
        arguments=["--master", "k8s://your_kubernetes_master_url",  # Replace with your actual master URL
                   "--deploy-mode", "cluster",
                   "--class", "org.apache.spark.examples.SparkPi",
                   "--conf", "spark.kubernetes.container.image=your_spark_image",
                   "--conf", "spark.kubernetes.namespace=your_namespace",
                   "--conf", "spark.kubernetes.authenticate.driver.serviceAccountName=spark",
                   "--conf", "spark.kubernetes.authenticate.executor.serviceAccountName=spark",
                   "--conf", "spark.driver.extraJavaOptions=-Dlog4j.configuration=log4j.properties",
                   "--conf", "spark.executor.extraJavaOptions=-Dlog4j.configuration=log4j.properties",
                   "local:///opt/spark/examples/jars/spark-examples_2.12-3.5.0.jar", # Replace with your Spark application jar
                   "--id", "{{ params.id }}"],
        kubernetes_conn_id='your_kubernetes_connection_id',
        params=arguments_to_pass,
        dag=dag,
    )

登录后复制

注意：

确保 namespace 参数设置为你的 Kubernetes 命名空间。
kubernetes_conn_id 参数设置为 Airflow 中配置的 Kubernetes 连接 ID。
这里使用了 KubernetesPodOperator，因为 SparkKubernetesOperator 不是标准的 Airflow Operator。你需要根据你的 Spark 集群配置调整 cmds 和 arguments 参数。特别注意 --master 的设置，确保其指向你的 Kubernetes 集群。

配置 SparkApplication YAML 文件 (或等效配置)

由于我们使用了 KubernetesPodOperator，不再直接使用 SparkApplication YAML 文件。参数通过 arguments 直接传递给 spark-submit 命令。

关键点：

在 arguments 中，使用 Jinja 模板 {{ params.id }} 来引用 Airflow DAG 中 params 字典中定义的参数。
确保参数名称与 Airflow DAG 中 params 字典中的键匹配。
通过 --id 这样的参数传递方式，使得参数能够被 spark-submit 命令正确识别。

在 Spark 应用程序中读取参数

在 Spark 应用程序中，可以使用 sys.argv 读取通过命令行传递的参数。

import sys

print("**********", sys.argv)

# 假设 'id' 参数是第一个参数
if len(sys.argv) > 1:
    id_value = sys.argv[sys.argv.index("--id") + 1] # 获取 --id 后的参数值
    print("ID:", id_value)
else:
    print("No arguments provided.")

登录后复制

注意：

sys.argv 是一个包含所有命令行参数的列表，包括脚本名称本身。
需要遍历 sys.argv 列表，找到对应的参数名称（例如 --id），然后获取其后的参数值。
进行适当的错误处理，以防止参数不存在或索引超出范围。

总结

通过以上步骤，可以确保在使用 Airflow 的 KubernetesPodOperator (或 SparkKubernetesOperator，如果可用) 时，能够正确地将参数传递给 Spark 应用程序。关键在于正确配置 Airflow DAG 中的 params 参数，并在 Spark 应用程序中使用 sys.argv 读取这些参数。同时，需要根据实际的 Spark 集群配置调整 KubernetesPodOperator 的 cmds 和 arguments 参数。

以上就是通过 Airflow SparkKubernetesOperator 传递参数的详细内容，更多请关注php中文网其它相关文章！