Airflow 時間變數的地雷區

Before 地雷

• start_date ⮕ DAG 開始的日期，必要參數
• end_date ⮕ DAG 停止的日期，不必要參數
• schedule_interval 或是 schedule ⮕ 每次執行的間隔，必要參數

排程三問

Q: 那還記得第一次執行的時間嗎？

A: 不是 start_date，而是 start_date + schedule_interval * 1

Q: 那第二次勒?

A: start_date + schedule_interval * 2

Q: 那我現在想執行勒？

A: 設定 start_date + schedule_interval * (1到n) = 現在，誒是數學課嗎🤣

坑的源頭：execution_date(已改名為logical_date)

execution_date 是什麼？

小西瓜：我知道，選我選我。
：好，小西瓜你回答～
小西瓜：execution_date 中文是執行時間，例如第一次執行的execution_date就是start_date + schedule_interval * 1
：登登登登～❌❌❌❌❌❌，答錯，因為...誒...其實我也覺得你的才對，但反正過去的定義不是這樣，execution_date 是說每次執行的開始時間，蛤～你說聽起來一樣啊，一圖千言，請看：

：每次執行都是一段時間的 DAG RUN，但是 DAG 不會在這段時間執行，會在結束之後才執行，DAG RUN 的時間就是 schedule_interval 的時間間隔長度
小西瓜：還是不懂
：沒關係 execution_date 已經棄用了，雖然還是可以使用，但官方不建議用
小西瓜：有沒有搞ㄘㄡ～～

logical_date 和 ds

雖然execution_date 是歷史遺留的產物，白話文就是技術債🥲，但是現在還是有這個變數，他改名了，以後叫做logical_date，所以學習他還是必須的，另外如果你只需要開始執行時間的日期，就可以用ds就可以了，白話文就是logical_date只拿日期這樣

你搞的我好亂啊！我需要更清楚的變數

data_interval_start = logical_date = execution_date

救世主來了，data_interval_start和 date_interval_end，清楚表達了 DAG RUN 的開始和結束，所以再問一次，每次的實際執行時間是誰？是date_interval_end喔～

有了這個以後我們就放棄 logical_date 和 execution_date 吧～管他現在叫什麼名字～

誒～那上次執行有成功嗎？

除了data_interval_start和 date_interval_end之外，我們也能夠利用
prev_data_interval_start_success、prev_data_interval_end_success、prev_start_date_success 這三個變數可以看過去的運行成功的日期

• prev_data_interval_start_success= 上次運行成功的 data_interval_start
• prev_data_interval_end_success= 上次運行成功的 date_interval_end
• prev_start_date_success= 上次運行成功的 ds

知道這些變數，然後呢？

使用執行時間來排程

例如你要抓每天的股票交易資料，可能是設定爬蟲在每天收盤後到證交所網站抓，突然你發現前幾天抓失敗了，這時候該怎麼做?
如果你程式沒有用時間變數，代表你可能是用 Python 的 datetime 模組來設定當天的時間，那你今天10/5要抓9/30、10/2、10/3的資料，只能手動了吧，或是有另外儲存錯誤的日期在另外抓資料進來執行，聽起來就很麻煩，如果你有用時間變數就很方便了，當然需要先進去爬坑🥲，但只要照常執行，給開始時間和結束時間，這時候給個參數設定只重跑error的，就沒問題了，因為當程式date_interval_end是10/2，就能抓到 10/2的資料了～

以為不是坑的坑：Timezone 時區設定

就說了不要改時區很難嗎～坑都是自己挖的，有時候會害到自己，大多時候會害到別人

你可以調整 Web UI 和程式當中的時區，這樣排程對時間會比較方便，但是程式實際執行時間都是UTC，所以有點像是快樂錶的概念，當我修改 Web UI 上的時區，我可以在 Web UI 看到 UTC+8，如下：

接著，也可以從 DAG 程式中設定 UTC+8 時區從10/1 開始的每天早上 9:00 執行(前提是已經度過上方的地雷區)，那他真的會在台灣時間每天早上 9:00 執行，例如明天早上 9:00就會執行，DAG 程式設定如下：

小提醒：如果設定是這樣，偵測到今天還沒執行的話，會直接run今天的部分，就算現在已經超過早上9點，但是dag run 執行以天為單位的話，是00:00，哎，這個坑先不展開了

import pendulum
from datetime import datetime

local_tz = pendulum.timezone("Asia/Taipei")

@dag(
    dag_id="test",
    schedule_interval="0 9 * * *",
    start_date = datetime(2023, 10, 1,tzinfo=local_tz),
)

但是，但是，這樣成功執行了沒錯，但是不管是 data_interval_start = logical_date = execution_date，ds等等變數，拿到的時間資料都是 UTC+0，??????，沒錯，就是這樣～以上方為例，明天就是會拿到

• data_interval_start = logical_date = execution_date ⮕ 10/5 1:00
• data_interval_end ⮕ 10/6 1:00
• ds ⮕ 10/5

誒～不懂嗎～代表你還在坑裡面

結語

時間變數和排程真的是很可怕的東西，哎，有哪些變數勒～
airflow 官方變數表格

要怎麼用

• 已經被棄用但還是抓得到的 execution_date

def get_date(execution_date=None):
    return execution_date

• 變數都在變數的神奇魔法 **context 裡，回到那一天 [Day10] Airflow Variable 變數的神秘魔法

def get_date(**context):
    return context['logical_date']

今天就先到這裡吧～