Airflow kubernetes pod operator env_vars when render_template_as_native_obj is True

176 Views Asked by At

In my Airflow DAG I needed to set render_template_as_native_obj=True to be able to customize the input value in one task. As far as I know is not possible to apply the flag only to the desired tasks, so this will affect all the tasks in the DAG.

My problem comes with env_vars of a Kubernetes Pod Operator. This is the task code, which was working before applying the flag:

download_missing_files = KubernetesPodOperator(
        task_id=f"download_missing_files_{file_to_load['task_alias']}",
        image="....",
        namespace='dbt',
        service_account_name='dbt-worker',
        name="googleads",
        get_logs=True,
        config_file="/home/airflow/composer_kube_config",
        kubernetes_conn_id='kubernetes_default',
        log_events_on_failure=True,
        env_vars={
            "missing_files": f"{list_missing_files.output['missing_files']}",
            "destination_bucket": f"{file_to_load['destination_bucket']}",
            "google_cred": f"{file_to_load['source_conn_id']}",
            "api_version": f"{file_to_load['g_ads_api_version']}",
            "backwards_days": "{{params.backwards_days}}",
            "data_owner_sa": f"{DATA_OWNER_SA}"
        },
        container_resources=k8s_models.V1ResourceRequirements(
            requests={"cpu": "1000m", "memory": "4G", "ephemeral-storage": "2G"},
            limits={"cpu": "1000m", "memory": "4G", "ephemeral-storage": "2G"},
        ),
        do_xcom_push=True,
    )

Afterwards I tried many different versions modifying the way env_var is created, but it keeps failing. This was one of the versions:

 env_vars={
    "missing_files": f"{{{{ ti.xcom_pull(key='manual_files_load_list', task_ids='list_missing_files_{file_to_load['task_alias']}) }}}}",
    "destination_bucket": f"{{{file_to_load['destination_bucket']}}}",
    "google_cred": f"{{{file_to_load['source_conn_id']}}}",
    "api_version": f"{{{file_to_load['g_ads_api_version']}}}",
    "backwards_days": "{{params.backwards_days}}",
    "data_owner_sa": f"{{{DATA_OWNER_SA}}}"
},

The error in this case:

jinja2.exceptions.TemplateSyntaxError: expected token 'end of print statement', got ':'

Does anyone understand how does it works? Thanks in advance.

0

There are 0 best solutions below