Airflow中SparkKubernetesOperator参数传递失败求助

核心问题排查与修正 #

1. 参数名拼写错误
   你在dag.py里写的parms={"ID": '1'}是拼写错误，Airflow Operator接收模板参数的正确字段是params，不是parms，这是导致参数无法传递的首要原因。

2. YAML模板的正确引用方式
   确保spark-pi.yaml中的arguments字段正确引用参数，同时要注意Spark的arguments格式为传递给主程序的参数数组，需按数组形式编写。

正确代码示例 #

1. dag.py 修正版 #

from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
}

with DAG(
    'spark_k8s_param_dag',
    default_args=default_args,
    description='Spark on K8s with parameter passing',
    schedule_interval=None,
    start_date=days_ago(1),
    tags=['spark', 'k8s'],
) as dag:

    spark_task = SparkKubernetesOperator(
        task_id='spark_submit',
        namespace='default',
        application_file='spark-pi.yaml',
        params={"ID": '1'},  # 修正为正确的params字段
        do_xcom_push=True,
        dag=dag,
    )

2. spark-pi.yaml 修正版 #

apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: "gcr.io/spark-operator/spark:v3.1.1"
  imagePullPolicy: Always
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.1.1.jar"
  arguments:
    - "{{ params.ID }}"  # 正确引用params中的ID参数
  sparkVersion: "3.1.1"
  restartPolicy:
    type: Never

3. processing.py 参数读取修正 #

如果是自定义Python脚本，需注意sys.argv[0]是脚本名，参数从sys.argv[1]开始读取：

import sys

if __name__ == "__main__":
    if len(sys.argv) > 1:
        id_param = sys.argv[1]
        print(f"Received ID parameter: {id_param}")
    else:
        print("No ID parameter received")

验证步骤 #

• 运行DAG后，查看Airflow任务日志，确认渲染后的SparkApplication YAML中arguments字段是否正确填充了1。
• 查看Spark Pod的日志，确认processing.py是否输出了正确的参数值。

内容的提问来源于stack exchange，提问作者Shivam Gupta