airflow - 如何在 Airflow 变量上强制 Jinja 模板？

Question

Airflow docs说:You can use Jinja templating with every parameter that is marked as “templated” in the documentation . Airflow 世界中的特定参数(例如 PythonOperator 的某些参数)由 Airflow 自动模板化是有道理的。我想知道获得非 Airflow 变量以进行模板化的最佳/正确方法是什么。我的具体用例类似于:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from somewhere import export_votes_data, export_queries_data
from elsewhere import ApiCaucus, ApiQueries

dag = DAG('export_training_data',
          description='Export training data for all active orgs to GCS',
          schedule_interval=None,
          start_date=datetime(2018, 3, 26), catchup=False)

HOST = "http://api-00a.dev0.solvvy.co"
BUCKET = "gcs://my-bucket-name/{{ ds }}/" # I'd like this to get templated

votes_api = ApiCaucus.get_votes_api(HOST)
queries_api = ApiQueries.get_queries_api(HOST)

export_votes = PythonOperator(task_id="export_votes", python_callable=export_votes_data,
                              op_args=[BUCKET, votes_api], dag=dag)
export_queries = PythonOperator(task_id="export_queries", python_callable=export_query_data,
                                op_args=[BUCKET, queries_api, export_solutions.task_id], dag=dag,
                                provide_context=True)

Answer 1

provide_context PythonOperator 的论据将传递用于模板的参数。来自 the documentation :

provide_context (bool) – if set to true, Airflow will pass a set of keyword arguments that can be used in your function. This set of kwargs correspond exactly to what you can use in your jinja templates. For this to work, you need to define **kwargs in your function header.

使用提供给可调用的上下文后，您可以在函数中进行插值:

def your_callable(bucket, api, **kwargs):
  bucket = bucket.format(**kwargs)
  [...]