Airflow中SparkSubmitOperator配置DAG级参数遇模板渲染异常问题

问题原因 #

SparkSubmitOperator默认仅对application_args、name、_application等少数字段启用Jinja模板渲染（这些字段被包含在Operator的template_fields列表中）。而executor_cores、executor_memory、driver_memory这类Spark资源配置参数不在默认的模板渲染范围内，因此直接写入{{ params.executor_cores }}时，Airflow不会解析模板语法，只会原样输出；若强行用双引号包裹，会触发DAG解析的语法错误导致无法启动。

解决方案 #

方案一：自定义SparkSubmitOperator扩展模板字段 #

通过继承原Operator并扩展template_fields，让目标参数支持模板渲染：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

class CustomSparkSubmitOperator(SparkSubmitOperator):
    # 在默认模板字段基础上，添加需要解析的Spark资源参数
    template_fields = SparkSubmitOperator.template_fields + ('executor_cores', 'executor_memory', 'driver_memory')

# 定义DAG
with DAG(
    dag_id="spark_param_dag",
    params={
        "executor_cores": "2",
        "executor_memory": "4g",
        "driver_memory": "2g",
        "application_args": ["--input-path", "/user/data/input"]
    },
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    spark_submit_task = CustomSparkSubmitOperator(
        task_id="run_spark_job",
        application="/path/to/your/spark_application.jar",
        executor_cores="{{ params.executor_cores }}",
        executor_memory="{{ params.executor_memory }}",
        driver_memory="{{ params.driver_memory }}",
        application_args="{{ params.application_args }}"
    )

方案二：用PythonOperator结合SparkSubmitHook实现 #

通过PythonCallable直接构建Spark提交参数，利用Airflow上下文获取DAG params，绕开原Operator的模板限制：

from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.hooks.spark_submit import SparkSubmitHook
from datetime import datetime

def submit_spark_job(**context):
    # 从上下文获取DAG级Params
    dag_params = context["params"]
    # 初始化SparkSubmitHook并传入解析后的参数
    spark_hook = SparkSubmitHook(
        application="/path/to/your/spark_application.jar",
        executor_cores=dag_params["executor_cores"],
        executor_memory=dag_params["executor_memory"],
        driver_memory=dag_params["driver_memory"],
        application_args=dag_params["application_args"]
    )
    # 执行Spark任务
    spark_hook.run()

# 定义DAG
with DAG(
    dag_id="spark_param_dag_python",
    params={
        "executor_cores": "2",
        "executor_memory": "4g",
        "driver_memory": "2g",
        "application_args": ["--input-path", "/user/data/input"]
    },
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    spark_submit_task = PythonOperator(
        task_id="run_spark_job",
        python_callable=submit_spark_job,
        provide_context=True
    )

注意事项 #

• 自定义Operator时，确保添加的字段是原SparkSubmitOperator的合法实例属性，避免出现属性不存在的错误
• 不要给Jinja模板语法额外添加双引号，Airflow会自动处理字符串类型参数的渲染逻辑
• 若使用Airflow 2.x+，PythonOperator的provide_context=True可替换为op_kwargs结合templates_dict，但直接使用上下文参数更直观

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