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

Đang mở Phù hợp với người mới
  • Có gì không ổn?
#75,762 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

@sdaberdaku đang làm issue này rồi.

Từ ngày 27/10/2026.

  • #75764 của @sdaberdaku — đang mở

Đánh giá

Độ khó
2/5
Thời gian dự kiến
1-3 giờ
Mức phù hợp với người mới
79/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
python
Lĩnh vực
backend, data-engineering

Hướng nghiên cứu

Bắt đầu với _PySparkDecoratedOperator và template_fields của nó, sau đó so sánh chúng với PySparkOperator.template_fields và mẫu toán tử Kubernetes được trang trí được đề cập trong issue. Tái hiện ví dụ bằng render_template_fields và xác nhận rằng conn_id và config_kwargs được render cho toán tử được trang trí giống như với toán tử cổ điển.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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
Ngôn ngữ chính
Python
Star
47.3k
Fork
18.2k
Merge trung bình
19 giờ 6 phút
Pull request đã merge (30 ngày)
651

Chuẩn bị môi trường

Mở trong Codespaces

Khởi chạy dev container của dự án ngay trên trình duyệt, bằng tài khoản GitHub của bạn.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/airflow

Tất cả issue của apache/airflow

Issue tương tự

Thêm issue về Python

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.