# ===== 基本的なリトライ設定 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta # デフォルト設定 default_args = { ‘owner’: ‘data_team’, ‘retries’: 3, # 失敗したら3回リトライ ‘retry_delay’: timedelta(minutes=5), # リトライ間隔は5分 ‘retry_exponential_backoff’: True, # 指数バックオフ ‘max_retry_delay’: timedelta(minutes=30) # 最大待機時間 } def my_task(): print(“タスク実行中…”) return “完了” with DAG( dag_id=’retry_example’, default_args=default_args, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: task = PythonOperator( task_id=’task_with_retry’, python_callable=my_task )

# ===== タスクごとに個別設定 ===== # 重要なタスクは多めにリトライ critical_task = PythonOperator( task_id=’critical_task’, python_callable=critical_function, retries=5, retry_delay=timedelta(minutes=2) ) # 重要度が低いタスクは少なめ non_critical_task = PythonOperator( task_id=’non_critical_task’, python_callable=non_critical_function, retries=1, retry_delay=timedelta(minutes=1) )

# ===== エラーの種類によってリトライを決める ===== from airflow.exceptions import AirflowException def smart_retry_function(): “””エラーの種類によってリトライを決める””” try: result = call_external_api() return result except ConnectionError as e: # 接続エラーはリトライする print(f”接続エラー: {e}”) raise # リトライさせる except ValueError as e: # データエラーはリトライしても無駄 print(f”データエラー: {e}”) raise AirflowException(f”リトライ不可能なエラー: {e}”) task = PythonOperator( task_id=’smart_retry_task’, python_callable=smart_retry_function, retries=3 )

# ===== タスクのタイムアウト ===== from datetime import timedelta task = PythonOperator( task_id=’task_with_timeout’, python_callable=long_running_function, execution_timeout=timedelta(minutes=30) # 30分でタイムアウト )

# ===== センサーのタイムアウト ===== from airflow.sensors.filesystem import FileSensor wait_for_file = FileSensor( task_id=’wait_for_file’, filepath=’/data/input.csv’, poke_interval=60, # 60秒ごとにチェック timeout=3600, # 1時間でタイムアウト mode=’poke’ )

# ===== メールによるエラー通知 ===== from airflow.utils.email import send_email def failure_callback(context): “””タスク失敗時のコールバック””” task_instance = context[‘task_instance’] exception = context.get(‘exception’) subject = f”[AIRFLOW ERROR] {task_instance.dag_id}.{task_instance.task_id}” html_content = f”””

タスクが失敗しました

項目	内容
DAG	{task_instance.dag_id}
タスク	{task_instance.task_id}
実行時刻	{context[‘execution_date’]}
試行回数	{task_instance.try_number}
エラー内容	{exception}

“”” send_email( to=’admin@example.com’, subject=subject, html_content=html_content ) # DAGに設定 default_args = { ‘on_failure_callback’: failure_callback }

# ===== Slackによるエラー通知 ===== from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook def slack_failure_callback(context): “””Slackにエラー通知””” task_instance = context[‘task_instance’] exception = context.get(‘exception’) slack_msg = f””” :x: *タスクが失敗しました* *DAG:* `{task_instance.dag_id}` *タスク:* `{task_instance.task_id}` *実行時刻:* {context[‘execution_date’]} *試行回数:* {task_instance.try_number} *エラー:* {exception} Airflow UIからログを確認してください。 “”” hook = SlackWebhookHook(slack_webhook_conn_id=’slack_webhook’) hook.send_text(slack_msg) task = PythonOperator( task_id=’task’, python_callable=some_function, on_failure_callback=slack_failure_callback )

# ===== 成功時の通知 ===== def success_callback(context): “””タスク成功時の通知””” task_instance = context[‘task_instance’] hook = SlackWebhookHook(slack_webhook_conn_id=’slack_webhook’) hook.send_text(f””” :white_check_mark: *タスクが正常に完了しました* *DAG:* `{task_instance.dag_id}` *タスク:* `{task_instance.task_id}` *実行時刻:* {context[‘execution_date’]} “””) task = PythonOperator( task_id=’task’, python_callable=some_function, on_success_callback=success_callback, on_failure_callback=slack_failure_callback )

# ===== ログの出力方法 ===== import logging def my_task(**context): logger = logging.getLogger(__name__) logger.info(“処理を開始します”) logger.debug(“デバッグ情報: データ件数 = 1000”) logger.warning(“警告: NULL値が含まれています”) logger.error(“エラー: データベース接続に失敗”) # 標準出力もログに記録される print(“標準出力もログに記録されます”) return “完了”

# ===== メトリクスの記録 ===== def process_data(**context): “””処理結果をメトリクスとして記録””” import pandas as pd df = pd.read_csv(‘/data/input.csv’) # 処理 processed_count = len(df) error_count = df[‘status’].value_counts().get(‘error’, 0) success_rate = (processed_count – error_count) / processed_count # メトリクスをXComに保存 context[‘ti’].xcom_push(key=’processed_count’, value=processed_count) context[‘ti’].xcom_push(key=’error_count’, value=error_count) context[‘ti’].xcom_push(key=’success_rate’, value=success_rate) print(f”処理件数: {processed_count}”) print(f”エラー件数: {error_count}”) print(f”成功率: {success_rate:.2%}”) return processed_count

# ===== べき等な処理の例 ===== def idempotent_load(**context): “””べき等な処理の例””” from sqlalchemy import create_engine import pandas as pd execution_date = context[‘ds’] engine = create_engine(‘postgresql://user:pass@localhost/db’) # 既存データを削除してから挿入（UPSERT） delete_query = f”DELETE FROM daily_summary WHERE date = ‘{execution_date}'” engine.execute(delete_query) # 新しいデータを挿入 df = pd.read_csv(f’/data/processed_{execution_date}.csv’) df.to_sql(‘daily_summary’, engine, if_exists=’append’, index=False) # 何度実行しても同じ結果になる！ print(f”✅ {execution_date}のデータをロード完了”)

# ===== 段階的な処理とチェックポイント ===== with DAG( dag_id=’robust_pipeline’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: # ステップ1: データ抽出 extract = PythonOperator( task_id=’extract_data’, python_callable=extract_data, retries=3 ) # チェックポイント1: 抽出データ検証 validate_extract = PythonOperator( task_id=’validate_extracted_data’, python_callable=validate_extracted_data ) # ステップ2: データ変換 transform = PythonOperator( task_id=’transform_data’, python_callable=transform_data, retries=2 ) # チェックポイント2: 変換データ検証 validate_transform = PythonOperator( task_id=’validate_transformed_data’, python_callable=validate_transformed_data ) # ステップ3: データロード load = PythonOperator( task_id=’load_data’, python_callable=load_data, retries=3 ) # 成功通知 notify = SlackWebhookOperator( task_id=’notify_success’, slack_webhook_conn_id=’slack_webhook’, message=”✅ パイプラインが正常に完了しました” ) # 依存関係 extract >> validate_extract >> transform >> validate_transform >> load >> notify

┌─────────┐     ┌──────────┐     ┌─────────┐     ┌──────────┐     ┌──────┐     ┌────────┐
│ Extract │ ──→ │ Validate │ ──→ │Transform│ ──→ │ Validate │ ──→ │ Load │ ──→ │ Notify │
│(retries │     │ Extract  │     │(retries │     │Transform │     │(retry│     │        │
│   =3)   │     │          │     │   =2)   │     │          │     │ =3)  │     │        │
└─────────┘     └──────────┘     └─────────┘     └──────────┘     └──────┘     └────────┘
     │               │                │               │              │
     ▼               ▼                ▼               ▼              ▼
   失敗時           失敗時            失敗時          失敗時         失敗時
   アラート         アラート          アラート        アラート       アラート
        

# ===== 堅牢なETLパイプライン ===== from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook from datetime import datetime, timedelta import pandas as pd import logging logger = logging.getLogger(__name__) def slack_alert(context, message_type=’failure’): “””Slack通知を送信””” ti = context[‘task_instance’] if message_type == ‘failure’: emoji = ‘:x:’ title = ‘タスクが失敗しました’ exception = context.get(‘exception’, ‘Unknown’) extra = f”*エラー:* {exception}” else: emoji = ‘:white_check_mark:’ title = ‘パイプラインが完了しました’ extra = “” msg = f””” {emoji} *{title}* *DAG:* `{ti.dag_id}` *タスク:* `{ti.task_id}` *実行日:* {context[‘ds’]} {extra} “”” hook = SlackWebhookHook(slack_webhook_conn_id=’slack_webhook’) hook.send_text(msg) def failure_callback(context): slack_alert(context, ‘failure’) def success_callback(context): slack_alert(context, ‘success’) def extract_data(**context): “””データ抽出””” logger.info(“データ抽出開始”) df = pd.DataFrame({ ‘id’: range(1, 101), ‘value’: [i * 10 for i in range(1, 101)] }) file_path = f”/tmp/extracted_{context[‘ds’]}.csv” df.to_csv(file_path, index=False) logger.info(f”✅ 抽出完了: {len(df)}件 → {file_path}”) return file_path def validate_data(**context): “””データ検証””” ti = context[‘ti’] file_path = ti.xcom_pull(task_ids=’extract’) logger.info(f”データ検証開始: {file_path}”) df = pd.read_csv(file_path) # 検証1: レコード数 if len(df) < 10: raise ValueError(f"レコード数が少なすぎます: {len(df)}") # 検証2: NULL値 null_count = df.isnull().sum().sum() if null_count > 0: logger.warning(f”NULL値が{null_count}件あります”) logger.info(f”✅ 検証完了: {len(df)}件”) return file_path def transform_data(**context): “””データ変換””” ti = context[‘ti’] file_path = ti.xcom_pull(task_ids=’validate’) logger.info(f”データ変換開始: {file_path}”) df = pd.read_csv(file_path) df[‘doubled’] = df[‘value’] * 2 output_path = f”/tmp/transformed_{context[‘ds’]}.csv” df.to_csv(output_path, index=False) logger.info(f”✅ 変換完了: {len(df)}件 → {output_path}”) return output_path def load_data(**context): “””データロード””” ti = context[‘ti’] file_path = ti.xcom_pull(task_ids=’transform’) logger.info(f”データロード開始: {file_path}”) df = pd.read_csv(file_path) # べき等性: 既存データを削除してから挿入 final_path = f”/tmp/final_{context[‘ds’]}.csv” df.to_csv(final_path, index=False) logger.info(f”✅ ロード完了: {len(df)}件 → {final_path}”) # メトリクスを記録 ti.xcom_push(key=’record_count’, value=len(df)) ti.xcom_push(key=’total_value’, value=int(df[‘doubled’].sum())) return final_path default_args = { ‘owner’: ‘data_team’, ‘retries’: 3, ‘retry_delay’: timedelta(minutes=5), ‘retry_exponential_backoff’: True, ‘on_failure_callback’: failure_callback } with DAG( dag_id=’robust_etl_pipeline’, default_args=default_args, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False, tags=[‘etl’, ‘robust’] ) as dag: extract = PythonOperator( task_id=’extract’, python_callable=extract_data, execution_timeout=timedelta(minutes=30) ) validate = PythonOperator( task_id=’validate’, python_callable=validate_data ) transform = PythonOperator( task_id=’transform’, python_callable=transform_data ) load = PythonOperator( task_id=’load’, python_callable=load_data, on_success_callback=success_callback ) extract >> validate >> transform >> load