在使用 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 参数。
