Jinja templating in Airflow can´t template class Object´s property

[MaksYermak]

Apache Airflow version

main (development)

What happened

When you will try to create an Object in operator properties and for this Object you will try to template his property in this case Jinja will not work.
Code example:

create_batch = DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_ID,
        region=REGION,
        batch=Batch(
            pyspark_batch=PySparkBatch(
                main_python_file_uri="main_python_file_uri",
                python_file_uris="files",
                args=[
                    "--start-date",
                    "{{ data_interval_start }}", # it doesn't work
                    "--end-date",
                    "{{ ds }}", # it doesn't work
                    "--table",
                    "table_name",
                    "--source-bucket",
                    "source_bucket",
                    "--source-prefix",
                    "source_prefix",
                    "--output-path",
                    "path",
                ],
            ),
        ),
        batch_id=f"pyspark-table-{str(uuid4())}",
    )

Result:

What you think should happen instead

No response

How to reproduce

For reproduce you can run this code

import os
from datetime import datetime
from uuid import uuid4

from airflow import models
from google.cloud.dataproc_v1 import Batch, PySparkBatch
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
)

PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "an-id")
REGION = os.environ.get("GCP_LOCATION", "europe-west1")

with models.DAG(
    "example_gcp_batch_dataproc",
    schedule_interval='@once',
    start_date=datetime(2021, 1, 1),
    catchup=False,
) as dag_batch:
    create_batch = DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_ID,
        region=REGION,
        batch=Batch(
            pyspark_batch=PySparkBatch(
                main_python_file_uri="main_python_file_uri",
                python_file_uris="files",
                args=[
                    "--start-date",
                    "{{ data_interval_start }}", # it doesn't work
                    "--end-date",
                    "{{ ds }}", # it doesn't work
                    "--table",
                    "table_name",
                    "--source-bucket",
                    "source_bucket",
                    "--source-prefix",
                    "source_prefix",
                    "--output-path",
                    "path",
                ],
            ),
        ),
        batch_id=f"pyspark-table-{str(uuid4())}",
    )

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[eladkal]

If you want to render Native Python object then you need to follow instructions on the docs and use render_template_as_native_obj=True:
https://airflow.apache.org/docs/apache-airflow/stable/concepts/operators.html#rendering-fields-as-native-python-objects

converting issue for github discussion as for the moment this doesn't appear to be a bug

[MaksYermak]

I have tried to add render_template_as_native_obj=True parameter to my code. Result was the same the problem still exist.

[alekscp]

I am experiencing the exact same issue with the DataprocCreateBatchOperator. Have you found a solution to your problem @MaksYermak ?

[MaksYermak]

@alekscp for fixing this problem you can try to use a dictionary instead of Batch and PySparkBatch objects.

[alekscp]

I just had a second look at your code example and just realized that you are using Batch and PySparkBatch. I am passing a dict to the batch property of the DataprocCreateBatchOperator and still experiencing the problem that you originally described. I've implemented a very hacky solution where I grab the values I need with a PythonOperator and store them in Airflow variables for my next operator to pick up. It's definitely not the recommended way to do things, but it works for now until I can fix those templating issues.