# ===== cron形式の構成 ===== # * * * * * # ┬ ┬ ┬ ┬ ┬ # │ │ │ │ └─ 曜日（0-6、0=日曜、7=日曜） # │ │ │ └─── 月（1-12） # │ │ └───── 日（1-31） # │ └─────── 時（0-23） # └───────── 分（0-59） # 【特殊文字】 # * : すべて # , : 複数指定（例: 1,3,5） # – : 範囲指定（例: 1-5） # / : 間隔指定（例: */15 = 15分ごと）

# ===== スケジュール設定の実装例 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def daily_report(): print(“日次レポートを作成中…”) # 例1: 毎日朝9時に実行 with DAG( dag_id=’daily_report_9am’, start_date=datetime(2024, 1, 1), schedule_interval=’0 9 * * *’, # 毎日9時0分 catchup=False ) as dag: task = PythonOperator( task_id=’create_report’, python_callable=daily_report ) # 例2: 15分ごとに実行 with DAG( dag_id=’every_15_minutes’, start_date=datetime(2024, 1, 1), schedule_interval=’*/15 * * * *’, # 15分ごと catchup=False ) as dag: task = PythonOperator( task_id=’check_data’, python_callable=lambda: print(“データチェック中…”) ) # 例3: 平日のみ8時に実行 with DAG( dag_id=’weekday_morning’, start_date=datetime(2024, 1, 1), schedule_interval=’0 8 * * 1-5′, # 月〜金 8時0分 catchup=False ) as dag: task = PythonOperator( task_id=’morning_task’, python_callable=lambda: print(“おはようございます”) )

# ===== timedeltaを使ったスケジュール ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta # 30分ごとに実行 with DAG( dag_id=’every_30_minutes’, start_date=datetime(2024, 1, 1), schedule_interval=timedelta(minutes=30), catchup=False ) as dag: task = PythonOperator( task_id=’my_task’, python_callable=lambda: print(“30分ごとの実行”) ) # 6時間ごとに実行 with DAG( dag_id=’every_6_hours’, start_date=datetime(2024, 1, 1), schedule_interval=timedelta(hours=6), catchup=False ) as dag: task = PythonOperator( task_id=’my_task’, python_callable=lambda: print(“6時間ごとの実行”) )

【execution_date = 処理対象のデータの日時】

タイムライン:
2024-01-01 00:00 ─────────────────────┐
     │                               │
     │  ← この日のデータを処理        │
     │                               │
2024-01-02 00:00 ←───── 実行時刻 ────┘
                        execution_date = 2024-01-01

なぜこうなる？
→ データは「過去のもの」を処理するから
→ 2024-01-01が終わってから、そのデータを処理する
        

# ===== execution_dateを使う ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def process_data(**context): # execution_dateを取得 execution_date = context[‘execution_date’] print(f”実行日時: {datetime.now()}”) print(f”処理対象日: {execution_date}”) # 日付をファイル名に使う filename = f”data_{execution_date.strftime(‘%Y%m%d’)}.csv” print(f”処理するファイル: {filename}”) with DAG( dag_id=’use_execution_date’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: task = PythonOperator( task_id=’process’, python_callable=process_data )

# ===== start_dateの設定方法 ===== # 方法1: 固定の日時を指定（推奨） from datetime import datetime start_date = datetime(2024, 1, 1) # 方法2: 現在から相対的に指定（非推奨） from datetime import datetime, timedelta start_date = datetime.now() – timedelta(days=1) # ❌ 避ける # 方法3: pendulumを使う（タイムゾーン対応、推奨） import pendulum start_date = pendulum.datetime(2024, 1, 1, tz=’Asia/Tokyo’)

【設定】
start_date = 2024-01-01
今日 = 2024-01-10
schedule_interval = @daily

【catchup=True の場合】
2024-01-02 ✅ 実行
2024-01-03 ✅ 実行
2024-01-04 ✅ 実行
2024-01-05 ✅ 実行
2024-01-06 ✅ 実行
2024-01-07 ✅ 実行
2024-01-08 ✅ 実行
2024-01-09 ✅ 実行
2024-01-10 ✅ 実行
→ 9日分すべて実行（バックフィル）

【catchup=False の場合】
2024-01-02 ⏭️ スキップ
2024-01-03 ⏭️ スキップ
    ...
2024-01-09 ⏭️ スキップ
2024-01-10 ✅ 実行
→ 最新の1回分だけ実行
        

# ===== catchupの設定 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def process(**context): execution_date = context[‘execution_date’] print(f”処理日: {execution_date}”) # catchup=False（推奨） with DAG( dag_id=’no_catchup’, start_date=datetime(2024, 1, 1), # 過去の日付でもOK schedule_interval=’@daily’, catchup=False # 過去分は実行しない ) as dag: task = PythonOperator( task_id=’process’, python_callable=process ) # catchup=True（過去分も実行） with DAG( dag_id=’with_catchup’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=True, # 過去分も実行する max_active_runs=3 # 同時実行は最大3つまで ) as dag: task = PythonOperator( task_id=’process’, python_callable=process )

# ===== REST APIからトリガー ===== import requests import json # Airflow API エンドポイント url = ‘http://localhost:8080/api/v1/dags/my_dag_id/dagRuns’ # 認証情報 auth = (‘airflow’, ‘airflow’) # リクエストボディ data = { “conf”: {}, “dag_run_id”: “manual_run_001” } # POSTリクエスト response = requests.post( url, auth=auth, headers={‘Content-Type’: ‘application/json’}, data=json.dumps(data) ) print(response.status_code) print(response.json())

# ===== 手動実行専用DAG ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime # 手動実行専用（自動実行しない） with DAG( dag_id=’manual_only’, start_date=datetime(2024, 1, 1), schedule_interval=None, # 手動実行のみ catchup=False ) as dag: task = PythonOperator( task_id=’manual_task’, python_callable=lambda: print(“手動で実行されました”) )

# ===== 毎日朝9時に実行される日次レポートDAG ===== # ファイル名: daily_report_9am.py from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta import pandas as pd import logging logger = logging.getLogger(__name__) # ===================================== # タスク関数の定義 # ===================================== def extract_yesterday_data(**context): “””前日のデータを抽出””” execution_date = context[‘execution_date’] target_date = execution_date.strftime(‘%Y-%m-%d’) logger.info(f”データ抽出開始: {target_date}”) # ダミーデータ（実際はDBやAPIから取得） data = { ‘日付’: [target_date] * 5, ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’, ‘いちご’], ‘売上’: [15000, 25000, 10000, 30000, 20000] } df = pd.DataFrame(data) temp_file = f’/tmp/extracted_{target_date}.csv’ df.to_csv(temp_file, index=False) logger.info(f”データ抽出完了: {len(df)}件”) return temp_file def transform_data(**context): “””データを変換””” ti = context[‘ti’] temp_file = ti.xcom_pull(task_ids=’extract’) logger.info(f”データ変換開始: {temp_file}”) df = pd.read_csv(temp_file) df[‘税込売上’] = (df[‘売上’] * 1.1).astype(int) df[‘手数料’] = (df[‘売上’] * 0.05).astype(int) df[‘純利益’] = df[‘税込売上’] – df[‘手数料’] execution_date = context[‘execution_date’] target_date = execution_date.strftime(‘%Y-%m-%d’) transformed_file = f’/tmp/transformed_{target_date}.csv’ df.to_csv(transformed_file, index=False) logger.info(f”データ変換完了: {len(df)}件”) return transformed_file def create_report(**context): “””レポートを作成””” ti = context[‘ti’] transformed_file = ti.xcom_pull(task_ids=’transform’) df = pd.read_csv(transformed_file) total_sales = df[‘売上’].sum() total_profit = df[‘純利益’].sum() execution_date = context[‘execution_date’] target_date = execution_date.strftime(‘%Y-%m-%d’) report_file = f’/tmp/daily_report_{target_date}.txt’ with open(report_file, ‘w’, encoding=’utf-8′) as f: f.write(f”日次レポート – {target_date}\n”) f.write(“=” * 50 + “\n\n”) f.write(f”総売上: ¥{total_sales:,}\n”) f.write(f”純利益: ¥{total_profit:,}\n”) f.write(f”利益率: {(total_profit/total_sales*100):.1f}%\n”) logger.info(f”レポート作成完了: {report_file}”) return report_file def notify_completion(**context): “””完了通知””” execution_date = context[‘execution_date’] target_date = execution_date.strftime(‘%Y-%m-%d’) logger.info(f”✅ {target_date}の日次レポートが完了しました”) # 実際はここでメールやSlackに通知 # ===================================== # DAGの定義 # ===================================== default_args = { ‘owner’: ‘data_team’, ‘depends_on_past’: False, ‘email’: [‘admin@example.com’], ‘email_on_failure’: True, ‘retries’: 2, ‘retry_delay’: timedelta(minutes=5) } with DAG( dag_id=’daily_report_9am’, default_args=default_args, description=’毎日朝9時に実行される日次レポート’, start_date=datetime(2024, 1, 1), schedule_interval=’0 9 * * *’, # 毎日9時0分 catchup=False, tags=[‘daily’, ‘report’] ) as dag: extract = PythonOperator( task_id=’extract’, python_callable=extract_yesterday_data ) transform = PythonOperator( task_id=’transform’, python_callable=transform_data ) report = PythonOperator( task_id=’report’, python_callable=create_report ) notify = PythonOperator( task_id=’notify’, python_callable=notify_completion ) extract >> transform >> report >> notify