# 作業ディレクトリを作成して移動 mkdir -p ~/docker-practice/step19 cd ~/docker-practice/step19 # 現在の場所を確認 pwd

# プロジェクトディレクトリを作成 mkdir -p ~/docker-practice/airflow-docker/{dags,logs,plugins} cd ~/docker-practice/airflow-docker # 権限設定（重要！） chmod -R 777 logs # 現在の場所を確認 pwd

# .envファイルを作成 cat > .env << ‘EOF’ # Airflow設定 AIRFLOW_UID=50000 AIRFLOW_GID=0 # PostgreSQL設定 POSTGRES_USER=airflow POSTGRES_PASSWORD=airflow POSTGRES_DB=airflow # Airflow管理者設定 _AIRFLOW_WWW_USER_USERNAME=admin _AIRFLOW_WWW_USER_PASSWORD=admin EOF # 内容を確認 cat .env

# docker-compose.ymlを作成 cat > docker-compose.yml << ‘EOF’ version: ‘3.8’ # 共通の環境変数 x-airflow-common: &airflow-common image: apache/airflow:2.7.0 environment: &airflow-common-env AIRFLOW__CORE__EXECUTOR: LocalExecutor AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow AIRFLOW__CORE__FERNET_KEY: ” AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: ‘true’ AIRFLOW__CORE__LOAD_EXAMPLES: ‘false’ AIRFLOW__WEBSERVER__SECRET_KEY: ‘your-secret-key-here’ volumes: – ./dags:/opt/airflow/dags – ./logs:/opt/airflow/logs – ./plugins:/opt/airflow/plugins user: “${AIRFLOW_UID:-50000}:0” depends_on: &airflow-common-depends-on postgres: condition: service_healthy services: # PostgreSQL（メタデータベース） postgres: image: postgres:13 environment: POSTGRES_USER: airflow POSTGRES_PASSWORD: airflow POSTGRES_DB: airflow volumes: – postgres_data:/var/lib/postgresql/data healthcheck: test: [“CMD”, “pg_isready”, “-U”, “airflow”] interval: 10s timeout: 5s retries: 5 start_period: 10s restart: unless-stopped # Airflow初期化（1回のみ実行） airflow-init: <<: *airflow-common entrypoint: /bin/bash command: – -c – | airflow db init airflow users create \ –username admin \ –firstname Admin \ –lastname User \ –role Admin \ –email admin@example.com \ –password admin echo “Airflow initialization completed!” environment: <<: *airflow-common-env restart: “no” # Airflow Webserver airflow-webserver: <<: *airflow-common command: webserver ports: – “8080:8080” healthcheck: test: [“CMD”, “curl”, “–fail”, “http://localhost:8080/health”] interval: 30s timeout: 10s retries: 5 start_period: 60s restart: unless-stopped depends_on: <<: *airflow-common-depends-on airflow-init: condition: service_completed_successfully # Airflow Scheduler airflow-scheduler: <<: *airflow-common command: scheduler healthcheck: test: [“CMD”, “curl”, “–fail”, “http://localhost:8974/health”] interval: 30s timeout: 10s retries: 5 start_period: 60s restart: unless-stopped depends_on: <<: *airflow-common-depends-on airflow-init: condition: service_completed_successfully volumes: postgres_data: EOF # 内容を確認 cat docker-compose.yml

# サンプルDAGを作成 cat > dags/sample_dag.py << ‘EOF’ from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator from datetime import datetime, timedelta # デフォルト引数 default_args = { ‘owner’: ‘airflow’, ‘depends_on_past’: False, ‘start_date’: datetime(2024, 1, 1), ‘email_on_failure’: False, ‘email_on_retry’: False, ‘retries’: 1, ‘retry_delay’: timedelta(minutes=5), } # タスク関数 def print_hello(): print(“Hello from Airflow!”) return “Hello World” def print_date(): current_date = datetime.now() print(f”Current date: {current_date}”) return str(current_date) def print_goodbye(): print(“Goodbye from Airflow!”) return “Goodbye” # DAG定義 with DAG( ‘sample_dag’, default_args=default_args, description=’A simple sample DAG’, schedule_interval=timedelta(days=1), catchup=False, tags=[‘example’, ‘sample’], ) as dag: # タスク1: Hello task_hello = PythonOperator( task_id=’print_hello’, python_callable=print_hello, ) # タスク2: 日付表示 task_date = PythonOperator( task_id=’print_date’, python_callable=print_date, ) # タスク3: Bashコマンド実行 task_bash = BashOperator( task_id=’run_bash’, bash_command=’echo “Running bash command!” && date’, ) # タスク4: Goodbye task_goodbye = PythonOperator( task_id=’print_goodbye’, python_callable=print_goodbye, ) # 依存関係: hello -> date -> bash -> goodbye task_hello >> task_date >> task_bash >> task_goodbye EOF # 内容を確認 cat dags/sample_dag.py

# ファイル構成を確認 find . -type f -name “*.py” -o -name “*.yml” -o -name “.env” | head -20

# 初期化と起動（初回は時間がかかります） docker-compose up -d # 起動状況を確認 docker-compose ps

# ログを確認（Ctrl+Cで終了） docker-compose logs -f airflow-webserver

# 停止 docker-compose down # ボリュームも削除（データが消える） docker-compose down -v

# PostgreSQL操作用DAGを作成 cat > dags/postgres_etl_dag.py << ‘EOF’ from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.postgres.operators.postgres import PostgresOperator from airflow.providers.postgres.hooks.postgres import PostgresHook from datetime import datetime, timedelta import random default_args = { ‘owner’: ‘airflow’, ‘start_date’: datetime(2024, 1, 1), ‘retries’: 1, ‘retry_delay’: timedelta(minutes=5), } def extract_data(**context): “””データ抽出（サンプルデータ生成）””” data = [] products = [‘ノートPC’, ‘マウス’, ‘キーボード’, ‘モニター’, ‘ヘッドセット’] for i in range(10): data.append({ ‘product’: random.choice(products), ‘amount’: random.randint(1000, 100000), ‘quantity’: random.randint(1, 10), }) # XComでデータを渡す context[‘ti’].xcom_push(key=’extracted_data’, value=data) print(f”Extracted {len(data)} records”) return data def load_data(**context): “””データをPostgreSQLに挿入””” # XComからデータを取得 data = context[‘ti’].xcom_pull(key=’extracted_data’, task_ids=’extract’) hook = PostgresHook(postgres_conn_id=’postgres_default’) for record in data: hook.run(f””” INSERT INTO sales (product, amount, quantity, sale_date) VALUES ( ‘{record[‘product’]}’, {record[‘amount’]}, {record[‘quantity’]}, CURRENT_DATE ); “””) print(f”Loaded {len(data)} records to PostgreSQL”) def check_data(): “””データ確認””” hook = PostgresHook(postgres_conn_id=’postgres_default’) records = hook.get_records(“SELECT COUNT(*) FROM sales;”) print(f”Total records in sales table: {records[0][0]}”) with DAG( ‘postgres_etl_dag’, default_args=default_args, description=’ETL pipeline with PostgreSQL’, schedule_interval=’@daily’, catchup=False, tags=[‘etl’, ‘postgres’], ) as dag: # テーブル作成 create_table = PostgresOperator( task_id=’create_table’, postgres_conn_id=’postgres_default’, sql=””” CREATE TABLE IF NOT EXISTS sales ( id SERIAL PRIMARY KEY, product VARCHAR(100), amount INTEGER, quantity INTEGER, sale_date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); “””, ) # データ抽出 extract = PythonOperator( task_id=’extract’, python_callable=extract_data, ) # データ読み込み load = PythonOperator( task_id=’load’, python_callable=load_data, ) # データ確認 check = PythonOperator( task_id=’check’, python_callable=check_data, ) create_table >> extract >> load >> check EOF # 内容を確認 head -30 dags/postgres_etl_dag.py

# コンテナの状態確認 docker-compose ps # 各サービスのログ確認 docker-compose logs postgres docker-compose logs airflow-webserver docker-compose logs airflow-scheduler # コンテナ内でコマンド実行 docker-compose exec airflow-webserver airflow dags list docker-compose exec airflow-webserver airflow tasks list sample_dag # データベース接続テスト docker-compose exec airflow-webserver airflow db check