# ===== 最小構成のDAG ===== # ファイル名: hello_dag.py # 配置場所: $AIRFLOW_HOME/dags/hello_dag.py from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime # 実行する関数 def hello(): print(“Hello, Airflow!”) # DAGの定義 dag = DAG( dag_id=’my_first_dag’, # DAGの名前（一意である必要がある） start_date=datetime(2024, 1, 1), # 開始日 schedule_interval=’@daily’, # 実行スケジュール catchup=False # 過去の実行をスキップ ) # タスクの定義 task = PythonOperator( task_id=’hello_task’, # タスクの名前 python_callable=hello, # 実行する関数 dag=dag # 所属するDAG )

# ===== PythonOperatorの基本 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract_data(): “””データを抽出する関数””” print(“データを抽出中…”) data = [1, 2, 3, 4, 5] print(f”データ抽出完了: {data}”) return data def transform_data(): “””データを変換する関数””” print(“データを変換中…”) print(“データ変換完了”) dag = DAG( dag_id=’python_operator_example’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) # タスク1: データ抽出 task1 = PythonOperator( task_id=’extract’, python_callable=extract_data, dag=dag ) # タスク2: データ変換 task2 = PythonOperator( task_id=’transform’, python_callable=transform_data, dag=dag ) # 依存関係: task1 → task2 task1 >> task2

# ===== BashOperatorの基本 ===== from airflow import DAG from airflow.operators.bash import BashOperator from datetime import datetime dag = DAG( dag_id=’bash_operator_example’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) # タスク1: ファイルを作成 task1 = BashOperator( task_id=’create_file’, bash_command=’echo “Hello, Airflow!” > /tmp/hello.txt’, dag=dag ) # タスク2: ファイルの内容を表示 task2 = BashOperator( task_id=’show_file’, bash_command=’cat /tmp/hello.txt’, dag=dag ) # タスク3: ファイルを削除 task3 = BashOperator( task_id=’delete_file’, bash_command=’rm /tmp/hello.txt’, dag=dag ) # 依存関係: task1 → task2 → task3 task1 >> task2 >> task3

# ===== 引数を渡す ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def greet(name, age): “””引数を受け取る関数””” print(f”こんにちは、{name}さん！”) print(f”あなたは{age}歳ですね。”) dag = DAG( dag_id=’with_arguments’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) task = PythonOperator( task_id=’greet_task’, python_callable=greet, op_kwargs={‘name’: ‘田中’, ‘age’: 30}, # 引数を辞書で渡す dag=dag )

【線形（順番に実行）】
A >> B >> C >> D

  A → B → C → D

【並列実行】
A >> [B, C, D]

      ┌→ B
  A → ┼→ C
      └→ D

【結合】
[A, B, C] >> D

  A →┐
  B →┼→ D
  C →┘

【分岐と結合】
A >> [B, C] >> D

      ┌→ B →┐
  A → ┤      ├→ D
      └→ C →┘
        

# ===== >> 演算子 ===== # task1 の後に task2 を実行 task1 >> task2 # task1 の後に task2、task2 の後に task3 を実行 task1 >> task2 >> task3 # task1 の後に task2 と task3 を並列実行 task1 >> [task2, task3] # task2 と task3 の後に task4 を実行 [task2, task3] >> task4

# ===== 複雑な依存関係 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def task_function(task_name): print(f”{task_name} を実行中…”) dag = DAG( dag_id=’complex_dependencies’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) # タスクを定義 start = PythonOperator( task_id=’start’, python_callable=task_function, op_kwargs={‘task_name’: ‘Start’}, dag=dag ) extract1 = PythonOperator( task_id=’extract1′, python_callable=task_function, op_kwargs={‘task_name’: ‘Extract1′}, dag=dag ) extract2 = PythonOperator( task_id=’extract2’, python_callable=task_function, op_kwargs={‘task_name’: ‘Extract2′}, dag=dag ) transform = PythonOperator( task_id=’transform’, python_callable=task_function, op_kwargs={‘task_name’: ‘Transform’}, dag=dag ) load = PythonOperator( task_id=’load’, python_callable=task_function, op_kwargs={‘task_name’: ‘Load’}, dag=dag ) end = PythonOperator( task_id=’end’, python_callable=task_function, op_kwargs={‘task_name’: ‘End’}, dag=dag ) # 依存関係を設定 # start # / \ # extract1 extract2 # \ / # transform # | # load # | # end start >> [extract1, extract2] >> transform >> load >> end

# ===== シンプルなETLパイプライン ===== # ファイル名: simple_etl_dag.py # 配置場所: $AIRFLOW_HOME/dags/simple_etl_dag.py from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta import pandas as pd import logging import os # ロガーを取得 logger = logging.getLogger(__name__) # ===================================== # ETL関数の定義 # ===================================== def extract_data(): “””データ抽出””” logger.info(“データ抽出を開始”) # サンプルデータを作成 data = { ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’], ‘価格’: [150, 100, 80, 300], ‘個数’: [10, 20, 15, 5] } df = pd.DataFrame(data) df.to_csv(‘/tmp/extracted_data.csv’, index=False) logger.info(f”データ抽出完了: {len(df)}件”) return ‘/tmp/extracted_data.csv’ def transform_data(): “””データ変換””” logger.info(“データ変換を開始”) df = pd.read_csv(‘/tmp/extracted_data.csv’) # 変換処理 df[‘税込価格’] = (df[‘価格’] * 1.1).astype(int) df[‘合計金額’] = df[‘税込価格’] * df[‘個数’] df.to_csv(‘/tmp/transformed_data.csv’, index=False) logger.info(f”データ変換完了: {len(df)}件”) return ‘/tmp/transformed_data.csv’ def load_data(): “””データロード””” logger.info(“データロードを開始”) df = pd.read_csv(‘/tmp/transformed_data.csv’) output_path = ‘/tmp/final_output.csv’ df.to_csv(output_path, index=False) logger.info(f”データロード完了: {output_path}”) logger.info(f”\n{df.to_string()}”) return output_path def cleanup(): “””クリーンアップ””” logger.info(“クリーンアップを開始”) files_to_delete = [ ‘/tmp/extracted_data.csv’, ‘/tmp/transformed_data.csv’ ] for file_path in files_to_delete: if os.path.exists(file_path): os.remove(file_path) logger.info(f”削除: {file_path}”) logger.info(“クリーンアップ完了”) # ===================================== # DAGの定義 # ===================================== default_args = { ‘owner’: ‘data_engineer’, ‘retries’: 2, ‘retry_delay’: timedelta(minutes=5) } dag = DAG( dag_id=’simple_etl_pipeline’, default_args=default_args, description=’シンプルなETLパイプライン’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False, tags=[‘etl’, ‘example’] ) # ===================================== # タスクの定義 # ===================================== extract_task = PythonOperator( task_id=’extract_data’, python_callable=extract_data, dag=dag ) transform_task = PythonOperator( task_id=’transform_data’, python_callable=transform_data, dag=dag ) load_task = PythonOperator( task_id=’load_data’, python_callable=load_data, dag=dag ) cleanup_task = PythonOperator( task_id=’cleanup’, python_callable=cleanup, dag=dag ) # ===================================== # 依存関係の設定 # ===================================== extract_task >> transform_task >> load_task >> cleanup_task

# ===== Context Managerを使った書き方 ===== from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract(): print(“Extract”) def transform(): print(“Transform”) def load(): print(“Load”) # Context Managerを使った書き方 with DAG( dag_id=’etl_with_context’, start_date=datetime(2024, 1, 1), schedule_interval=’@daily’, catchup=False ) as dag: # タスク定義（dagパラメータ不要！） extract_task = PythonOperator( task_id=’extract’, python_callable=extract ) transform_task = PythonOperator( task_id=’transform’, python_callable=transform ) load_task = PythonOperator( task_id=’load’, python_callable=load ) # 依存関係 extract_task >> transform_task >> load_task