Airflowのマクロがうまく動かない場合に確認すること

背景

以前こんな感じの質問を受けた。

{{}} の中に書けるマクロがあるかと思いますが、書いてもプリプロセスしてくれなくて。
何かおまじないや、設定追加が必要なものなのでしょうか。

これに関してメモ。

まず、Airflowのマクロとは

Airflow Document# Macros reference

Airflowのマクロとはテンプレートで挙動する変数
例えば、 {{ execution_date }} を呼び出すと、実行時間をPendulumクラスのインスタンスで取得できる。

動く場合と動かない場合

ドキュメントにもあるが、このMacro は Jinja Templates内で有効なものである。

例を示す。

動く例↓

動く例

t1 = BigQueryOperator(
    task_id="example",
    sql="./sql/example.sql",
    use_legacy_sql=False,
    destination_dataset_table="pj.mart.example${{ execution_date.strftime('%Y%m%d') }}", # イケる！
    write_disposition="WRITE_TRUNCATE",
    allow_large_results=True,
    time_partitioning={"type": "DAY"},
    dag=dag
)

動かない例↓

動かない例

ds = {{execution_date}}.to_date_string # ここでエラー
BigQueryOperator(~略~)

Operatorの、とある引数フィールドであればMacroは使用することができる。

じゃあ、どのOperatorのどの引数でmacro使えるのか？

Jinja Templating

You can use Jinja templating with every parameter that is marked as “templated” in the documentation.

ドキュメント見て、 templated って書いてある項目ならイケるらしい。

例えば、

airflow.contrib.operators.bigquery_operator

をみると、sql や destination_dataset_table に templated と書いてあるので macroを使用できることがわかる。