AirflowでSensorOperator使うときのexecution_timeout, retries, timeout, poke_intervalの使い方

# TL;DR
- `BaseOperator.execution_timeout` も `SensorOperator.timeout` も同様の挙動をする
  - 汎用的な `BaseOperator.execution_timeout` を使うことがおすすめ
- タスクが失敗(タイムアウト/例外)になった場合に、 `retry_delay` の間隔で `retries` の回数再実行される 
- `SensorOperator` では、処理がpassするまで `poke_interval` の間隔で処理が何度も実行される
  - 処理が `passしない` ことと、処理が `失敗する` ことを分けて考える必要がある
     - 処理がpassしない -> 処理の失敗ではない。`poke_interval` 毎に処理が実行されるだけ
     - 処理が例外/タイムアウト -> 処理の失敗。retryの対象となる

# 背景
`BaseOperator` のコンストラクタ引数には `execution_timeout` と `retries` がある。これは、タスクのタイムアウト時間と再実行回数を指定できる。

一方、 `xxxSensorOperator` のコンストラクタ引数には `timeout` と`poke_inverval` が用意されている。

- `execution_timeout` と `timeout` の違いは何か
- SensorOperatorの処理がうまくいかなかった場合に、`poke_interval` と `retries` の違いは？
  - `poke_interval` と `retry_delay` の違いは？

など不明が多いのでまとめる。


# 1. BaseOperatorの execution_timeout と retriesの動きを確認

シンプルにexecution_timeout時刻でタスクがfailとなり、retriesの回数再実行される。想定通り。


```py

import airflow
from datetime import timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import time

dag = DAG("base_dag")

def sleep_py():
    print('sleep')
    time.sleep(30)
    return 'success'

sp = PythonOperator(
    task_id='sleep_py',
    python_callable=sleep_py,
    execution_timeout=timedelta(seconds=3),
    retries=3,
    retry_delay=timedelta(seconds=1),
    start_date=airflow.utils.dates.days_ago(1),
    dag=dag,
)

# "sleep" が3回出力される

```

# 2. SensorOperator(return false)で timeout と poke_interval の動きを確認

poke_intervalの間隔で処理が何度も実行され、timeoutの秒数経過するとタスクがfailとなる。こちらも想定通り。

```py
import airflow
from airflow.models import DAG
from airflow.utils.dates import days_ago
from datetime import timedelta
from airflow.contrib.sensors.python_sensor import PythonSensor

def false_py():
    print("false_py")
    return False

dag = DAG("sensor_test", schedule_interval="@daily")

sensor = PythonSensor(
        task_id='sensor_4',
        poke_interval=9,
        timeout=60,
        python_callable=false_py,
        start_date=airflow.utils.dates.days_ago(1),
        dag=dag
        )

# false_pyが5回出力される
# false_pyがFalseを返すためSensorの処理がパスしないから(Trueでパス)
```

# 3. execution_timeout と timeout を同時に使うとどうなるか

指定秒数が短い方が有効となる。以下の場合だと、短いexecution_timeoutの30秒でタイムアウトとなる。

```py

import airflow
from airflow.models import DAG
from airflow.utils.dates import days_ago
from datetime import timedelta
from airflow.contrib.sensors.python_sensor import PythonSensor

def false_py():
    print("false_py")
    return False

dag = DAG("sensor_test", schedule_interval="@daily")

sensor = PythonSensor(
        task_id='sensor_5',
        poke_interval=9,
        timeout=60,
        execution_timeout=timedelta(seconds=30),
        python_callable=false_py,
        start_date=airflow.utils.dates.days_ago(1),
        dag=dag
        )
# false_py が2回出力される
```

# 4. SensorOperator(return false)に execution_timeout, poke_interval, retries を指定するとどうなるか

poke_intervalの間隔で処理が実行され、execution_timeoutの秒数経過するとタスクfailとなり、retriesで指定された回数再実行される。

```py

import airflow
from airflow.models import DAG
from airflow.utils.dates import days_ago
from datetime import timedelta
from airflow.contrib.sensors.python_sensor import PythonSensor

def false_py():
    print("false_py")
    return False

dag = DAG("sensor_test", schedule_interval="@daily")

sensor = PythonSensor(
        task_id='sensor_7',
        poke_interval=9,
        execution_timeout=timedelta(seconds=60),
        python_callable=false_py,
        start_date=airflow.utils.dates.days_ago(1),
        retries=3,
        retry_delay=timedelta(seconds=1),
        dag=dag
        )

# false_py が5回出力される ← 3回実行される
```

# 5. SensorOperatorに poke_interval と retries が使われている場合に、例外処理するとどうなるか

初回の処理実行で例外となり、retriesの回数だけタスクが再実行されて終わる。

```py

import airflow
from airflow.models import DAG
from airflow.utils.dates import days_ago
from datetime import timedelta
from airflow.contrib.sensors.python_sensor import PythonSensor

def fail_py():
    1/0
    print("fail_py")
    return False

dag = DAG("sensor_test", schedule_interval="@daily")

sensor = PythonSensor(
        task_id='sensor_7',
        poke_interval=10,
        execution_timeout=timedelta(seconds=60),
        python_callable=fail_py,
        start_date=airflow.utils.dates.days_ago(1),
        retries=3,
        retry_delay=timedelta(seconds=1),
        dag=dag
        )

# 例外が3回飛ぶ
```