【STEP 20で学んだ関係と同じパターン】 Apache Beam（レシピ）→ Dataflow（調理場） Apache Airflow（楽譜）→ Cloud Composer（コンサートホール） ┌─────────────────────────────────────────────────────┐ │ Apache Airflow（オープンソース） │ │ ├── Pythonでワークフローを定義 │ │ ├── DAG（有向非巡回グラフ）で処理を表現 │ │ └── 実行環境: │ │ ├── Cloud Composer ← 今回学ぶ（GCP） │ │ ├── MWAA（AWS版のマネージドAirflow） │ │ ├── Astronomer（専門サービス） │ │ └── 自分でサーバー構築（大変！） │ └─────────────────────────────────────────────────────┘ 【Cloud Composerを使うメリット】 ・Airflowの面倒な設定・運用をGoogleが全部やってくれる ・GCPサービス（BigQuery、Dataflow等）との連携が簡単 ・Webブラウザで状態監視、ログ確認できる

【default_argsの全パラメータを理解しよう】 default_args = { ‘owner’: ‘airflow’, ‘depends_on_past’: False, ‘start_date’: datetime(2024, 1, 1), ‘email_on_failure’: False, ‘email_on_retry’: False, ‘retries’: 1, ‘retry_delay’: timedelta(minutes=5), } ┌─────────────────────────────────────────────────────────┐ │ ‘owner’: ‘airflow’ │ ├─────────────────────────────────────────────────────────┤ │ DAGの所有者（担当者名） │ │ ・誰が管理しているかを明示 │ │ ・’data-team’、’tanaka’ など任意の名前 │ │ ・Airflow UIで担当者でフィルタリング可能 │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ ‘depends_on_past’: False │ ├─────────────────────────────────────────────────────────┤ │ 前回の実行結果に依存するか │ │ ・False → 前回が失敗でも今回は実行する │ │ ・True → 前回が成功していないと今回は実行しない │ │ ※ 通常はFalseで問題なし │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ ‘start_date’: datetime(2024, 1, 1) │ ├─────────────────────────────────────────────────────────┤ │ DAGの開始日（スケジュールの基準日） │ │ ・この日以降からスケジュールが有効になる │ │ ・過去の日付を指定するのが一般的 │ │ ※ 未来の日付だと、その日まで実行されない │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ ‘email_on_failure’: False │ ├─────────────────────────────────────────────────────────┤ │ タスク失敗時にメール通知するか │ │ ・True → 失敗したらメール送信 │ │ ・’email’: [‘alert@example.com’] と併用 │ │ ※ SMTPサーバーの設定が別途必要 │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ ‘retries’: 1 │ ├─────────────────────────────────────────────────────────┤ │ 失敗時の再試行回数 │ │ ・1 → 失敗したら1回だけ再試行 │ │ ・3 → 最大3回まで再試行 │ │ ・0 → 再試行しない │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ ‘retry_delay’: timedelta(minutes=5) │ ├─────────────────────────────────────────────────────────┤ │ 再試行までの待ち時間 │ │ ・timedelta(minutes=5) → 5分後に再試行 │ │ ・timedelta(seconds=30) → 30秒後に再試行 │ │ ・API制限などを考慮して適切な間隔を設定 │ └─────────────────────────────────────────────────────────┘

【DAG定義の各パラメータを理解しよう】 dag = DAG( ‘hello_world’, default_args=default_args, description=’最初のDAG’, schedule_interval=’@daily’, catchup=False, ) ┌─────────────────────────────────────────────────────────┐ │ ‘hello_world’ （第1引数） │ ├─────────────────────────────────────────────────────────┤ │ DAG ID（ユニークな識別子） │ │ ・Airflow UI に表示される名前 │ │ ・英数字とアンダースコアのみ推奨 │ │ ・例：’daily_sales_report’, ‘hourly_etl_job’ │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ schedule_interval=’@daily’ │ ├─────────────────────────────────────────────────────────┤ │ 実行スケジュール │ │ ・’@daily’ → 毎日0時0分 │ │ ・’@hourly’ → 毎時0分 │ │ ・’0 9 * * *’ → 毎日9時0分（Cron形式） │ │ ・None → 手動実行のみ │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ catchup=False │ ├─────────────────────────────────────────────────────────┤ │ 過去分の実行をするか（重要！） │ │ │ │ 【例】start_date=2024/1/1、今日=2024/1/15 の場合 │ │ │ │ catchup=True: │ │ 1/1, 1/2, 1/3, … 1/15 の15日分を全部実行 │ │ → 過去分を一気に処理（バックフィル） │ │ │ │ catchup=False: │ │ 1/15 だけ実行 │ │ → 過去分は無視（通常はこちら） │ │ │ │ ※ 新規DAGでは catchup=False を推奨！ │ │ Trueだと大量の過去分が実行されて大変なことに… │ └─────────────────────────────────────────────────────────┘

【configuration辞書の構造】 configuration={ ‘query’: { ‘query’: ‘SELECT …’, # 実行するSQL ‘useLegacySql’: False, # 標準SQL使用（必ずFalse） ‘destinationTable’: {…}, # 結果の出力先（省略可） ‘writeDisposition’: ‘…’ # 書き込みモード } } 【writeDispositionの選択肢】 ・’WRITE_TRUNCATE’ → 既存データを削除して上書き ・’WRITE_APPEND’ → 既存データに追加 ・’WRITE_EMPTY’ → テーブルが空の時のみ書込

【XComの基本操作】 ① データを保存する（push） context[‘ti’].xcom_push(key=’キー名’, value=データ) ・key: データを識別する名前（文字列） ・value: 保存するデータ（JSON化可能なもの） ② データを取得する（pull） data = context[‘ti’].xcom_pull( key=’キー名’, task_ids=’データを保存したタスクID’ ) 【注意点】 ⚠️ 大きなデータ（数MB以上）は保存しない！ → XComはメタデータDB（Cloud SQL）に保存される → 大きなデータはGCSに保存してパスだけXComで渡す ⚠️ **context が必要 → PythonOperatorで **context を受け取る必要がある → python_callable=関数名 だけでOK（自動で渡される） 【良い例と悪い例】 ❌ 悪い例：大きなDataFrameを直接渡す context[‘ti’].xcom_push(key=’data’, value=huge_dataframe) ✅ 良い例：ファイルパスを渡す huge_dataframe.to_csv(‘gs://bucket/data.csv’) context[‘ti’].xcom_push(key=’path’, value=’gs://bucket/data.csv’)

【Cron形式の書き方】 ‘分 時 日 月 曜日’ │ │ │ │ │ │ │ │ │ └── 曜日（0-6、0=日曜） │ │ │ └────── 月（1-12） │ │ └───────── 日（1-31） │ └──────────── 時（0-23） └─────────────── 分（0-59） 【特殊文字】 * = すべての値（毎〜） , = 複数指定（1,15 = 1日と15日） – = 範囲指定（1-5 = 1〜5） / = 間隔指定（*/15 = 15分ごと） 【よく使う例】 ‘0 9 * * *’ → 毎日9時0分 ’30 14 * * *’ → 毎日14時30分 ‘0 */6 * * *’ → 6時間ごと（0時、6時、12時、18時） ‘0 9 * * 1’ → 毎週月曜日9時0分 ‘0 9 * * 1-5’ → 平日（月〜金）9時0分 ‘0 0 1 * *’ → 毎月1日0時0分 ‘0 9 1,15 * *’ → 毎月1日と15日の9時0分 ‘*/15 * * * *’ → 15分ごと ‘0 2 * * *’ → 毎日午前2時（バッチ処理に最適） 【タイムゾーンに注意！】 Airflowのデフォルトは UTC です。 日本時間（JST）は UTC+9 なので： 日本時間9時に実行したい場合： → UTC 0時 = ‘0 0 * * *’ 日本時間18時に実行したい場合： → UTC 9時 = ‘0 9 * * *’