`@task.pyspark` does not template `config_kwargs` and `conn_id`

未关闭 适合新手
  • 哪里不对？
#75,762 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

@sdaberdaku 已经在做这个了。

开始于 2026年10月27日。

  • #75764 来自 @sdaberdaku —— 未关闭

评估

难度
2/5
预计耗时
1-3 小时
新手友好度
79/100
Issue 类型
缺陷
描述清晰度
描述清楚
活跃度
活跃
技术栈
python
领域
backend, data-engineering

调研方向

从 _PySparkDecoratedOperator 及其 template_fields 入手，然后将它们与 PySparkOperator.template_fields 以及 issue 中提到的 Kubernetes 装饰器运算符模式进行比较。使用 render_template_fields 重现示例，并确认装饰器运算符的 conn_id 和 config_kwargs 能像经典运算符那样进行渲染。

由索引模型根据 Issue 内容生成。

描述

needs-triage
Under which category would you file this issue?

Providers

Apache Airflow version

3.3.2

What happened and how to reproduce it?

@task.pyspark does not render Jinja in config_kwargs (or conn_id). The same arguments passed to PySparkOperator are rendered.

PySparkOperator declares both as template fields:

PySparkOperator.template_fields == ("conn_id", "config_kwargs", "templates_dict", "op_args", "op_kwargs")

_PySparkDecoratedOperator does not define template_fields. DecoratedOperator comes before PySparkOperator in its MRO, so it inherits DecoratedOperator's fields and drops the two Spark-specific ones:

_PySparkDecoratedOperator.template_fields == ("op_args", "op_kwargs")

Templated values in config_kwargs therefore reach Spark as literal text when the decorator is used.

Reproduction:

from airflow.providers.apache.spark.operators.spark_pyspark import PySparkOperator
from airflow.sdk import DAG, task

conf = {"spark.app.name": "app-{{ run_id }}"}

with DAG("repro", schedule=None):

    @task.pyspark(config_kwargs=conf)
    def decorated(spark=None):
        pass

    decorated_op = decorated().operator
    classic_op = PySparkOperator(task_id="classic", python_callable=lambda spark=None: None, config_kwargs=conf)

context = {"run_id": "manual__1"}
for op in (decorated_op, classic_op):
    op.render_template_fields(context)
    print(type(op).__name__, op.config_kwargs["spark.app.name"])

Output:

_PySparkDecoratedOperator app-{{ run_id }}
PySparkOperator app-manual__1
What you think should happen instead?

@task.pyspark should template conn_id and config_kwargs exactly like PySparkOperator. The decorated operator can declare the union of both sets of fields, the way _KubernetesDecoratedOperator keeps the fields of KubernetesPodOperator:

template_fields: Sequence[str] = (*DecoratedOperator.template_fields, "conn_id", "config_kwargs")
Operating System

Linux

Deployment

Virtualenv installation

Apache Airflow Provider(s)

apache-spark

Versions of Apache Airflow Providers

apache-airflow-providers-apache-spark==6.3.4
apache-airflow-providers-common-compat==1.23.0

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

The decorator never templated these fields: before PySparkOperator was introduced it declared template_fields = ("op_args", "op_kwargs") explicitly. The operator refactor made the inconsistency visible, since the same arguments now behave differently on the two entry points.

Are you willing to submit PR?
  • Yes I am willing to submit a PR!
Code of Conduct
主要语言
Python
星标
47.3k
派生
18.2k
平均合并
19 小时 6 分钟
30 天内合并 PR
651

环境准备

在 Codespaces 中打开

在浏览器里用你自己的 GitHub 账号启动这个项目的开发容器。

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

apache/airflow 的其他 Issue

查看 apache/airflow 的全部 Issue

相似的 Issue

更多 Python Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。