Airflow資料抄訳(16)：オペレーター（Operators）



恥ずかしながら最近になって知ったワークフローエンジン [Apache Airflow](https://airflow.apache.org/)。日本語の紹介記事もちらほら出てきていますが、[公式ドキュメント](https://airflow.apache.org/docs/)をちょっとずつ抄訳しながら読んでいこうと思います。

16回目の今回は[オペレーター](https://airflow.apache.org/docs/apache-airflow/stable/concepts/operators.html)（Operators）。
バージョン2.3.3時点のものです。

---

# オペレーター（Operators）

DAG:オペレーターはあらかじめ定義された[タスク](https://airflow.apache.org/docs/apache-airflow/stable/concepts/tasks.html)のためのテンプレートです。オペレーターを使用することでDAGの内部で宣言的にタスクを定義することが出来ます：

```py
with DAG("my-dag") as dag:
    ping = SimpleHttpOperator(endpoint="http://example.com/update/")
    email = EmailOperator(to="admin@example.com", subject="Update complete")

    ping >> email
```

Airflowにはとてもたくさんのオペレーターが用意されています。それらはAirflowのコアが組み込みで提供するものやプリインストールされたプロバイダーが提供するものです。コアにはいくつかの汎用性の高いオペレーターが含まれています：

* `BashOperator `- bashでコマンドを実行する
* `PythonOperator` - 任意のPython関数を実行する
* `EmailOperator` - Eメールを送信する

コア・オペレーターの一覧はこちらをご参照ください： [コア・オペレーターとフックのリファレンス](https://airflow.apache.org/docs/apache-airflow/stable/operators-and-hooks-ref.html)。

デフォルトでインストールされているものの中に必要なオペレーターがない場合でも、コミュニティが提供する膨大な[プロバイダー・パッケージ群](https://airflow.apache.org/docs/apache-airflow-providers/index.html)の中で見つかるかもしれません：

* `SimpleHttpOperator`
* `MySqlOperator`
* `PostgresOperator`
* `MsSqlOperator`
* `OracleOperator`
* `JdbcOperator`
* `DockerOperator`
* `HiveOperator`
* `S3FileTransformOperator`
* `PrestoToMySqlOperator`
* `SlackAPIOperator`

こうして示したもの以外にもたくさんのオペレーターが提供されています。コミュニティで管理されているオペレーター、フック、センサー、そしてトランスファー（transfer）が[プロバイダー・パッケージ群](https://airflow.apache.org/docs/apache-airflow-providers/operators-and-hooks-ref/index.html)のドキュメントで参照できます。

:::note warn
Note
Airflowのコード内では、[タスク](https://airflow.apache.org/docs/apache-airflow/stable/concepts/tasks.html)とオペレーターの概念はしばしば互換のものとして用いられています。タスクはDAGを構成する「実行の最小単位」の総称です。オペレーターは再利用可能で、事前定義されたタスク・テンプレートです。ユーザー開発者が作成したロジックを実行します。いくつかの引数（argument）を必要とします。
:::

## Jinjaテンプレート

Airflowは[Jinjaテンプレート・エンジン](http://jinja.pocoo.org/docs/dev/)を利用しており、[マクロ](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#templates-ref)と組み合わせて用いることでとてもパワフルに機能します。

例えば、データ区間（Data Interval）の開始日時を、環境変数を通じて` BashOperator `により実行されるbashスクリプトに渡したい場合は：

```py
# データ区間の開始日時。書式はYYYY-MM-DD
date = "{{ ds }}"
t = BashOperator(
    task_id="test_env",
    bash_command="/tmp/test.sh ",
    dag=dag,
    env={"DATA_INTERVAL_START": date},
)
```

ここで、`{{ ds }}`はテンプレート変数であり、`BashOperator`の`env`パラメーターはJinjaによりテンプレート化されているので、bashスクリプトは`DATA_INTERVAL_START`という名前の環境変数でこのデータ区間の開始日時にアクセスできます。

Jinjaテンプレート・エンジンが利用できるパラメーターはドキュメント上で「テンプレート化されている」（tepmlated）と記述されています。テンプレート・エンジンによる文字列置換は`pre_execute`関数が実行される直前に行われます。

Jinjaテンプレート・エンジンは入れ子のフィールドにも使えます。そのためには当該の入れ子のフィールドが「テンプレート化されている」ものであることを構造的に示す必要があります。`template_fields`プロパティに登録されているフィールドはテンプレート・エンジンによる文字列置換の対象になります。下記の例では`path`フィールドが該当します：

```py
class MyDataReader:
    template_fields: Sequence[str] = ("path",)

    def __init__(self, my_path):
        self.path = my_path

    # ここにその他のコード


t = PythonOperator(
    task_id="transform_data",
    python_callable=transform_data,
    op_args=[MyDataReader("/tmp/{{ ds }}/my_file")],
    dag=dag,
)
```

:::note warn
Note
`template_fields`プロパティはクラス変数でもインスタンス変数でも同様に機能します。
:::

深く入れ子になったフィールドもまた文字列置換の対象にできます。この場合入れ子関係上中間に位置するフィールドもまたテンプレート化されている必要があります：

```py
class MyDataTransformer:
    template_fields: Sequence[str] = ("reader",)

    def __init__(self, my_reader):
        self.reader = my_reader

    # ここにその他のコード


class MyDataReader:
    template_fields: Sequence[str] = ("path",)

    def __init__(self, my_path):
        self.path = my_path

    # ここにその他のコード


t = PythonOperator(
    task_id="transform_data",
    python_callable=transform_data,
    op_args=[MyDataTransformer(MyDataReader("/tmp/{{ ds }}/my_file"))],
    dag=dag,
)
```

DAGの作成時にJinjaの`Environment`に対してオプションを渡すことが出来ます。よくある使用例の1つは、テンプレート文字列の終端の改行文字を削除しないようJinjaに指示するというものです：

```py
my_dag = DAG(
    dag_id="my-dag",
    jinja_environment_kwargs={
        "keep_trailing_newline": True,
        # ここにその他のjinja2 Environmentオプション
    },
)
```

指定可能なオプションについて詳しくは[Jinjaのドキュメント](https://jinja.palletsprojects.com/en/2.11.x/api/#jinja2.Environment)を参照してください。

### ネイティブのPythonオブジェクトとしてフィールドをレンダリングする

デフォルトでは、`template_fields`に列挙されたフィールドの値はすべて文字列としてレンダリングされます。

例えば`extract`という名前のタスクが辞書（例：`{"1001": 301.27, "1002": 433.21, "1003": 502.22}`）を[XCom](https://airflow.apache.org/docs/apache-airflow/stable/concepts/xcoms.html#concepts-xcom)に登録したとします。後続タスクが実行されるとき、その値を取り出し文字列としてレンダリングされた結果を`order_data`引数に渡します（例：`'{"1001": 301.27, "1002": 433.21, "1003": 502.22}'`）。

```py
transform = PythonOperator(
    task_id="transform",
    op_kwargs={"order_data": "{{ti.xcom_pull('extract')}}"},
    python_callable=transform,
)
```

レンダリング結果としてネイティブのPythonオブジェクト（この例では`dict`オブジェクト）を得たい場合、DAGのパラメーターとして`render_template_as_native_obj=True`を指定します：

```py
dag = DAG(
    dag_id="example_template_as_python_object",
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    render_template_as_native_obj=True,
)


def extract():
    data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
    return json.loads(data_string)


def transform(order_data):
    print(type(order_data))
    for value in order_data.values():
        total_order_value += value
    return {"total_order_value": total_order_value}


extract_task = PythonOperator(task_id="extract", python_callable=extract)

transform_task = PythonOperator(
    task_id="transform",
    op_kwargs={"order_data": "{{ti.xcom_pull('extract')}}"},
    python_callable=transform,
)

extract_task >> transform_task
```

この例では、`order_data`引数に`{"1001": 301.27, "1002": 433.21, "1003": 502.22}`が渡されます。

`render_template_as_native_obj`に`True`が指定された場合、AirflowはJinjaの[NativeEnvironment](https://jinja.palletsprojects.com/en/2.11.x/nativetypes/)を利用します。`NativeEnvironment`により、テンプレートのレンダリング結果としてネイティブのPythonデータ型のインスタンスが生成されます。

## 予約語params

Apache Airflow v2.2.0において、`params`という変数名はDAGシリアライゼーション過程で利用されます。この変数名をサードパーティのオペレーター内で利用しないでください。以前のバージョンのAirflowからアップグレードをした場合、次のようなエラーに遭遇するかもしれません：

```
AttributeError: 'str' object has no attribute '__module__'
```

オペレーター内で使用している`params`変数の名前を変更してください。

---

レディメイドのオペレーターでは不足で、あるシステムに固有の、しかし当該システム内ではそれなりに汎用性のあるロジックを独自のオペレーターとして定義する場合、Jinjaを使ったパラメータ化が行えるのは便利そうだなと感じました。

一方で、「ネイティブのPythonオブジェクトとしてフィールドをレンダリングする」（Rendering Fields as Native Python Objects）のセクションで説明されている機能については、正直何がうれしいのだろう？と感じてしまいました。テンプレートのレンダリング結果として文字列以外のオブジェクトを想定するようなケースとは・・・と。