# ===== XComの基本 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def push_data(**context): “””データをプッシュ（送信）””” # データを返すと自動的にXComにプッシュされる return “Hello from task1” def pull_data(**context): “””データをプル（取得）””” ti = context[‘ti’] # TaskInstance # 前のタスクからデータを取得 data = ti.xcom_pull(task_ids=’push_task’) print(f”受信したデータ: {data}”) with DAG( dag_id=’xcom_example’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: task1 = PythonOperator( task_id=’push_task’, python_callable=push_data ) task2 = PythonOperator( task_id=’pull_task’, python_callable=pull_data ) task1 >> task2

# ===== 様々なデータ型をXComで渡す ===== def push_various_data(**context): “””様々なデータ型をプッシュ””” ti = context[‘ti’] # 文字列 ti.xcom_push(key=’string_data’, value=’Hello’) # 数値 ti.xcom_push(key=’number_data’, value=123) # リスト ti.xcom_push(key=’list_data’, value=[1, 2, 3, 4, 5]) # 辞書 ti.xcom_push(key=’dict_data’, value={ ‘name’: ‘田中’, ‘age’: 30, ‘city’: ‘東京’ }) # デフォルトのreturn値もプッシュされる return “All data pushed” def pull_various_data(**context): “””様々なデータ型をプル””” ti = context[‘ti’] # キーを指定してプル string_val = ti.xcom_pull(task_ids=’push_task’, key=’string_data’) number_val = ti.xcom_pull(task_ids=’push_task’, key=’number_data’) list_val = ti.xcom_pull(task_ids=’push_task’, key=’list_data’) dict_val = ti.xcom_pull(task_ids=’push_task’, key=’dict_data’) # キーを指定しない場合はreturn値を取得 default_val = ti.xcom_pull(task_ids=’push_task’) print(f”String: {string_val}”) print(f”Number: {number_val}”) print(f”List: {list_val}”) print(f”Dict: {dict_val}”) print(f”Default: {default_val}”)

# ===== ファイルパスの受け渡し ===== import pandas as pd def extract_data(**context): “””データを抽出してファイルパスを返す””” df = pd.DataFrame({ ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘価格’: [150, 100, 80] }) file_path = ‘/tmp/extracted_data.csv’ df.to_csv(file_path, index=False) print(f”データを保存: {file_path}”) # ファイルパスを返す（XComに自動プッシュ） return file_path def transform_data(**context): “””ファイルを読み込んで変換””” ti = context[‘ti’] # 前のタスクからファイルパスを取得 file_path = ti.xcom_pull(task_ids=’extract’) print(f”ファイルを読み込み: {file_path}”) df = pd.read_csv(file_path) df[‘税込価格’] = (df[‘価格’] * 1.1).astype(int) output_path = ‘/tmp/transformed_data.csv’ df.to_csv(output_path, index=False) print(f”変換後データを保存: {output_path}”) return output_path with DAG( dag_id=’file_path_xcom’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: extract = PythonOperator( task_id=’extract’, python_callable=extract_data ) transform = PythonOperator( task_id=’transform’, python_callable=transform_data ) extract >> transform

# ===== 従来の方法 vs TaskFlow API ===== # ❌ 従来の方法（冗長） def extract(**context): ti = context[‘ti’] data = “some data” ti.xcom_push(key=’data’, value=data) def transform(**context): ti = context[‘ti’] data = ti.xcom_pull(task_ids=’extract’, key=’data’) # 処理… # ✅ TaskFlow API（シンプル） from airflow.decorators import task @task def extract(): return “some data” # 自動でXComにプッシュ @task def transform(data): # 引数で自動的に受け取り # 処理… pass # 実行（関数呼び出しのように書ける） data = extract() transform(data)

# ===== TaskFlow APIの基本 ===== from airflow.decorators import dag, task from datetime import datetime @dag( start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) def taskflow_example(): @task def extract(): “””データ抽出””” data = [1, 2, 3, 4, 5] print(f”抽出: {data}”) return data @task def transform(data): “””データ変換””” transformed = [x * 2 for x in data] print(f”変換: {transformed}”) return transformed @task def load(data): “””データロード””” print(f”ロード: {data}”) total = sum(data) print(f”合計: {total}”) return total # DAGの実行フロー（関数呼び出しのように書ける） extracted_data = extract() transformed_data = transform(extracted_data) load(transformed_data) # DAGを登録 dag_instance = taskflow_example()

# ===== TaskFlow APIを使ったETLパイプライン ===== from airflow.decorators import dag, task from datetime import datetime import pandas as pd @dag( start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False, tags=[‘etl’, ‘taskflow’] ) def etl_with_taskflow(): “””TaskFlow APIを使ったETLパイプライン””” @task def extract() -> str: “””データ抽出””” df = pd.DataFrame({ ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’], ‘価格’: [150, 100, 80, 300], ‘個数’: [10, 20, 15, 5] }) file_path = ‘/tmp/extracted.csv’ df.to_csv(file_path, index=False) print(f”✅ データ抽出: {len(df)}件”) return file_path @task def transform(file_path: str) -> str: “””データ変換””” df = pd.read_csv(file_path) df[‘税込価格’] = (df[‘価格’] * 1.1).astype(int) df[‘合計金額’] = df[‘税込価格’] * df[‘個数’] output_path = ‘/tmp/transformed.csv’ df.to_csv(output_path, index=False) print(f”✅ データ変換: {len(df)}件”) return output_path @task def load(file_path: str) -> dict: “””データロード””” df = pd.read_csv(file_path) final_path = ‘/tmp/final_output.csv’ df.to_csv(final_path, index=False) summary = { ‘件数’: len(df), ‘総売上’: int(df[‘合計金額’].sum()), ‘ファイル’: final_path } print(f”✅ データロード: {summary}”) return summary # ETLフローの実行 extracted = extract() transformed = transform(extracted) result = load(transformed) # DAGを登録 dag_instance = etl_with_taskflow()

# ===== Web UIでの設定方法 ===== # 1. Web UI (http://localhost:8080) にアクセス # 2. 上部メニュー「Admin」→「Variables」をクリック # 3. 「+」ボタンで新規作成 # 4. Keyとvalueを入力 # 例: Key=”s3_bucket”, Val=”my-data-bucket” # 5. 「Save」をクリック

# ===== PythonでVariableを使う ===== from airflow.models import Variable from airflow.decorators import dag, task from datetime import datetime @dag( start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) def use_variables(): @task def get_config(): “””Variableから設定を取得””” # Variableを取得 s3_bucket = Variable.get(‘s3_bucket’) api_url = Variable.get(‘api_url’) print(f”S3バケット: {s3_bucket}”) print(f”API URL: {api_url}”) # デフォルト値を指定 timeout = Variable.get(‘timeout’, default_var=30) print(f”タイムアウト: {timeout}秒”) get_config() dag_instance = use_variables()

# ===== JSONでVariableを保存 ===== from airflow.models import Variable # Variable（Web UIで設定）: # Key: “app_config” # Val: {“database”: {“host”: “localhost”, “port”: 5432}, “s3”: {“bucket”: “my-bucket”}} @task def use_json_variable(): # JSONとして取得 config = Variable.get(‘app_config’, deserialize_json=True) print(f”DB Host: {config[‘database’][‘host’]}”) print(f”DB Port: {config[‘database’][‘port’]}”) print(f”S3 Bucket: {config[‘s3’][‘bucket’]}”)

# ===== Connectionを使ってDBに接続 ===== from airflow.hooks.base import BaseHook from airflow.decorators import dag, task from datetime import datetime import pandas as pd @dag( start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) def use_connection(): @task def query_database(): “””Connectionを使ってDBに接続””” # Connectionを取得 conn = BaseHook.get_connection(‘my_postgres’) # 接続文字列を作成 connection_string = ( f”postgresql://{conn.login}:{conn.password}” f”@{conn.host}:{conn.port}/{conn.schema}” ) print(f”接続先: {conn.host}:{conn.port}/{conn.schema}”) # SQLAlchemyで接続 from sqlalchemy import create_engine engine = create_engine(connection_string) # SQLクエリを実行 df = pd.read_sql(“SELECT * FROM users LIMIT 10″, engine) print(f”取得件数: {len(df)}”) return len(df) query_database() dag_instance = use_connection()

# ===== 完全なETLパイプライン ===== from airflow.decorators import dag, task from airflow.models import Variable from datetime import datetime, timedelta import pandas as pd import logging logger = logging.getLogger(__name__) @dag( dag_id=’complete_etl_pipeline’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False, tags=[‘etl’, ‘taskflow’, ‘complete’] ) def complete_etl_pipeline(): “””Variables + TaskFlow APIを使った完全なETLパイプライン””” @task def get_config() -> dict: “””Variablesから設定を取得””” config = { ‘output_dir’: Variable.get(‘output_dir’, default_var=’/tmp’), ‘tax_rate’: float(Variable.get(‘tax_rate’, default_var=’0.1′)) } logger.info(f”設定を取得: {config}”) return config @task def extract(config: dict) -> str: “””データ抽出””” df = pd.DataFrame({ ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’, ‘いちご’], ‘価格’: [150, 100, 80, 300, 250], ‘個数’: [10, 20, 15, 5, 8] }) file_path = f”{config[‘output_dir’]}/extracted.csv” df.to_csv(file_path, index=False) logger.info(f”✅ Extract完了: {len(df)}件 → {file_path}”) return file_path @task def transform(file_path: str, config: dict) -> str: “””データ変換””” df = pd.read_csv(file_path) tax_rate = config[‘tax_rate’] df[‘税込価格’] = (df[‘価格’] * (1 + tax_rate)).astype(int) df[‘合計金額’] = df[‘税込価格’] * df[‘個数’] output_path = f”{config[‘output_dir’]}/transformed.csv” df.to_csv(output_path, index=False) logger.info(f”✅ Transform完了: {len(df)}件 → {output_path}”) return output_path @task def load(file_path: str, config: dict) -> dict: “””データロード””” df = pd.read_csv(file_path) final_path = f”{config[‘output_dir’]}/final_report.csv” df.to_csv(final_path, index=False) summary = { ‘件数’: len(df), ‘総売上’: int(df[‘合計金額’].sum()), ‘平均単価’: int(df[‘価格’].mean()), ‘ファイル’: final_path } logger.info(f”✅ Load完了: {summary}”) return summary @task def notify(summary: dict): “””完了通知””” logger.info(“=” * 50) logger.info(“📊 ETLパイプライン完了レポート”) logger.info(“=” * 50) logger.info(f” 処理件数: {summary[‘件数’]}件”) logger.info(f” 総売上: ¥{summary[‘総売上’]:,}”) logger.info(f” 平均単価: ¥{summary[‘平均単価’]:,}”) logger.info(f” 出力先: {summary[‘ファイル’]}”) logger.info(“=” * 50) # ETLフローの実行 config = get_config() extracted = extract(config) transformed = transform(extracted, config) summary = load(transformed, config) notify(summary) # DAGを登録 dag_instance = complete_etl_pipeline()