# ===== PythonOperatorの基本 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime # 実行したい関数を定義 def hello_world(): print(“Hello, Airflow!”) return “タスク完了” # DAGの定義 with DAG( dag_id=’python_operator_example’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: # PythonOperatorでタスクを作成 task = PythonOperator( task_id=’hello_task’, python_callable=hello_world # 実行する関数 )

# ===== 引数を渡す ===== def greet(name, age): print(f”こんにちは、{name}さん！{age}歳ですね。”) return f”{name}さんの処理完了” # op_kwargsで引数を渡す task = PythonOperator( task_id=’greet_task’, python_callable=greet, op_kwargs={ ‘name’: ‘太郎’, ‘age’: 25 } )

# ===== contextを使って実行情報を取得 ===== def process_with_context(**context): “””contextから実行情報を取得””” # 実行日（execution_date） execution_date = context[‘ds’] print(f”実行日: {execution_date}”) # DAG ID dag_id = context[‘dag’].dag_id print(f”DAG ID: {dag_id}”) # タスクID task_id = context[‘task_instance’].task_id print(f”タスク ID: {task_id}”) # XComにデータを保存 context[‘ti’].xcom_push(key=’status’, value=’success’) return “処理完了” task = PythonOperator( task_id=’context_task’, python_callable=process_with_context, provide_context=True # Airflow 2.0以降は自動でTrueになる )

# ===== データベースからデータ抽出 ===== import pandas as pd from sqlalchemy import create_engine def extract_data(**context): “””データベースからデータを抽出””” engine = create_engine(‘postgresql://user:pass@localhost/mydb’) # 実行日を使ってクエリ execution_date = context[‘ds’] query = f””” SELECT * FROM sales WHERE date = ‘{execution_date}’ “”” df = pd.read_sql(query, engine) print(f”抽出件数: {len(df)}”) # CSVに保存 file_path = f’/tmp/sales_{execution_date}.csv’ df.to_csv(file_path, index=False) return file_path extract_task = PythonOperator( task_id=’extract_sales_data’, python_callable=extract_data )

# ===== BashOperatorの基本 ===== from airflow.operators.bash import BashOperator # 単純なコマンド task1 = BashOperator( task_id=’print_date’, bash_command=’date’ ) # 複数のコマンドを実行 task2 = BashOperator( task_id=’create_and_list’, bash_command=’mkdir -p /tmp/airflow_test && ls -la /tmp/airflow_test’ )

# ===== Airflowマクロを使う ===== task = BashOperator( task_id=’use_macros’, bash_command=”’ echo “実行日: {{ ds }}” echo “実行日（フォーマット）: {{ ds_nodash }}” echo “前日: {{ yesterday_ds }}” echo “翌日: {{ tomorrow_ds }}” ”’ )

# ===== 環境変数を渡す ===== task = BashOperator( task_id=’use_env_var’, bash_command=’echo “DB: $DB_NAME, ENV: $ENVIRONMENT”‘, env={ ‘DB_NAME’: ‘production’, ‘ENVIRONMENT’: ‘prod’ } )

# ===== ファイルバックアップ ===== backup_task = BashOperator( task_id=’backup_files’, bash_command=”’ # バックアップディレクトリ作成 BACKUP_DIR=”/backup/{{ ds }}” mkdir -p $BACKUP_DIR # ファイルをコピー cp -r /data/important_files $BACKUP_DIR/ # 圧縮 tar -czf $BACKUP_DIR.tar.gz $BACKUP_DIR # 古いバックアップを削除（30日以前） find /backup -name “*.tar.gz” -mtime +30 -delete echo “✅ バックアップ完了: $BACKUP_DIR.tar.gz” ”’ )

# ===== EmailOperatorの基本 ===== from airflow.operators.email import EmailOperator send_email = EmailOperator( task_id=’send_email’, to=’recipient@example.com’, subject=’Airflowからのお知らせ’, html_content=”’

タスクが完了しました！

処理が正常に終了しました。

”’ )

# ===== 動的な内容 ===== send_email = EmailOperator( task_id=’send_report’, to=’team@example.com’, subject=’日次レポート – {{ ds }}’, html_content=”’

日次レポート

実行日: {{ ds }}

DAG: {{ dag.dag_id }}

ステータス: 正常終了

”’ )

# ===== 添付ファイル付きメール ===== send_email = EmailOperator( task_id=’send_with_attachment’, to=’manager@example.com’, subject=’レポート添付 – {{ ds }}’, html_content=’

レポートを添付しました。ご確認ください。

‘, files=[‘/tmp/report.csv’, ‘/tmp/summary.pdf’] )

# ===== SlackWebhookOperatorの基本 ===== from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator send_slack = SlackWebhookOperator( task_id=’send_slack’, slack_webhook_conn_id=’slack_webhook’, message=’タスクが完了しました！’, channel=’#data-pipeline’ )

# ===== リッチメッセージ ===== send_slack = SlackWebhookOperator( task_id=’send_rich_message’, slack_webhook_conn_id=’slack_webhook’, message=”’ :white_check_mark: *データパイプライン完了* *DAG:* `sales_daily_pipeline` *実行日:* {{ ds }} *ステータス:* 成功 処理件数: 10,000件 処理時間: 5分30秒 ”’, channel=’#data-pipeline’ )

# ===== カスタムOperatorの基本 ===== from airflow.models import BaseOperator class MyCustomOperator(BaseOperator): “””カスタムOperatorの例””” def __init__(self, my_param, *args, **kwargs): super().__init__(*args, **kwargs) self.my_param = my_param def execute(self, context): “””実際の処理を書く””” self.log.info(f”パラメータ: {self.my_param}”) self.log.info(“カスタム処理を実行中…”) # 処理を実装 result = self.my_param * 2 self.log.info(f”結果: {result}”) return result # 使い方 task = MyCustomOperator( task_id=’custom_task’, my_param=10 )

# ===== CSVファイル検証Operator ===== import pandas as pd from airflow.models import BaseOperator class CsvValidationOperator(BaseOperator): “””CSVファイルを検証するカスタムOperator””” def __init__( self, file_path, required_columns=None, max_null_ratio=0.1, *args, **kwargs ): super().__init__(*args, **kwargs) self.file_path = file_path self.required_columns = required_columns or [] self.max_null_ratio = max_null_ratio def execute(self, context): “””検証処理””” self.log.info(f”ファイル検証開始: {self.file_path}”) # CSVを読み込み df = pd.read_csv(self.file_path) # 1. 必須カラムの存在確認 missing = set(self.required_columns) – set(df.columns) if missing: raise ValueError(f”必須カラムが不足: {missing}”) # 2. NULL値の割合チェック for col in df.columns: null_ratio = df[col].isnull().sum() / len(df) if null_ratio > self.max_null_ratio: self.log.warning(f”⚠️ ‘{col}’ のNULL値が多い: {null_ratio:.2%}”) self.log.info(f”✅ 検証完了: {len(df)}行”) return { ‘row_count’: len(df), ‘column_count’: len(df.columns), ‘validation_passed’: True } # 使い方 validate_csv = CsvValidationOperator( task_id=’validate_sales_csv’, file_path=’/data/sales.csv’, required_columns=[‘date’, ‘product_id’, ‘amount’], max_null_ratio=0.05 )

# ===== 複数Operatorを組み合わせたDAG ===== from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator from airflow.operators.email import EmailOperator from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator from datetime import datetime import pandas as pd def extract_data(**context): “””データ抽出””” df = pd.DataFrame({ ‘日付’: [‘2024-01-15’] * 100, ‘商品’: [f’商品{i}’ for i in range(100)], ‘売上’: [i * 100 for i in range(100)] }) file_path = f”/tmp/sales_{context[‘ds’]}.csv” df.to_csv(file_path, index=False) print(f”✅ 抽出完了: {len(df)}件 → {file_path}”) # 次のタスクにパスを渡す context[‘ti’].xcom_push(key=’file_path’, value=file_path) context[‘ti’].xcom_push(key=’record_count’, value=len(df)) return file_path with DAG( dag_id=’multi_operator_dag’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False, tags=[‘example’, ‘multi-operator’] ) as dag: # 1. データ抽出（Python） extract = PythonOperator( task_id=’extract_data’, python_callable=extract_data ) # 2. ファイル圧縮（Bash） compress = BashOperator( task_id=’compress_file’, bash_command=”’ FILE=”{{ ti.xcom_pull(task_ids=’extract_data’, key=’file_path’) }}” gzip -f $FILE echo “✅ 圧縮完了: ${FILE}.gz” ”’ ) # 3. Slack通知 notify_slack = SlackWebhookOperator( task_id=’notify_slack’, slack_webhook_conn_id=’slack_webhook’, message=”’ :white_check_mark: *日次データ処理完了* *実行日:* {{ ds }} *処理件数:* {{ ti.xcom_pull(task_ids=’extract_data’, key=’record_count’) }}件 ”’ ) # 4. メール送信 send_email = EmailOperator( task_id=’send_email’, to=’team@example.com’, subject=’日次データ処理完了 – {{ ds }}’, html_content=”’

処理完了レポート

日次データ処理が正常に完了しました。

処理件数: {{ ti.xcom_pull(task_ids=’extract_data’, key=’record_count’) }}件

”’ ) # 依存関係: extract → compress → [slack, email] extract >> compress >> [notify_slack, send_email]