主页 / coding / 问题 / 79594354
Accepted
Shweta Dalal
Asked: 2025-04-27 03:58:49 +0800 CST

为什么当我将 Airflow Dag docker 化时它会运行两次？

  • 772

我使用 Airflow 的 dag 进行了 docker 化，它基本上会将一个 csv 文件复制到 GCP，然后将其加载到 BigQuery 并进行简单的转换。当我运行 docker-compose run 时，我的 dag 会被执行两次。我无法理解这是由哪部分代码引起的。当我从 UI 手动触发 dag 时，它只运行一次。

我的达格：

from scripts import extract_and_gcpload, load_to_BQ

default_args = {
    'owner': 'shweta',
    'start_date': datetime(2025, 4, 24),
    'retries': 0
}

with DAG(
    'spacex_etl_dag',
    default_args=default_args,
    schedule_interval=None,
    schedule=None,
    catchup=False              #prevents Airflow from running missed periods
) as dag:

    extract_and_upload = PythonOperator(
        task_id="extract_and_upload_to_gcs",
        python_callable=extract_and_gcpload.load_to_gcp_pipeline,
    )

    load_to_bq = PythonOperator(
        task_id="load_to_BQ",
        python_callable=load_to_BQ.load_csv_to_bigquery
    )

    run_dbt = BashOperator(
        task_id="run_dbt",
        bash_command="cd '/opt/airflow/dbt/my_dbt' && dbt run --profiles-dir /opt/airflow/dbt"
    )

    extract_and_upload >> load_to_bq >> run_dbt

我的入口点文件startscript.sh：

#!/bin/bash
set -euo pipefail

log() {
  echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}

# Optional: Run DB init & parse DAGs
log "Initializing Airflow DB..."
airflow db upgrade

log "Parsing DAGs..."
airflow scheduler --num-runs 1

DAG_ID="spacex_etl_dag"
log "Unpausing DAG: $DAG_ID"
airflow dags unpause "$DAG_ID" || true

log "Triggering DAG: $DAG_ID"
airflow dags trigger "$DAG_ID" || true

log "Creating admin user (if not exists)..."
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email [email protected] \
    --password admin || true

if [[ "$1" == "webserver" || "$1" == "scheduler" ]]; then
  log "Starting Airflow: $1"
  exec airflow "$@"
else
  log "Executing: $@"
  exec "$@"
fi

我的 docker-compose.yaml 文件：

services:

  airflow-webserver:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: airflow-webserver
    env_file: .env
    restart: always
    environment:
      AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'false'
      AIRFLOW__LOGGING__REMOTE_LOGGING: 'False'
      AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      GOOGLE_APPLICATION_CREDENTIALS: /opt/airflow/secrets/llms-395417-c18ea70a3f54.json
    volumes:
      - ./dags:/opt/airflow/dags
      - ./scripts:/opt/airflow/scripts
      - ./dbt:/opt/airflow/dbt
      - ./secrets:/opt/airflow/secrets
    ports:
      - 8080:8080
    command: webserver

  airflow-scheduler:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: airflow-scheduler
    env_file: .env
    restart: always
    environment:
      AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'false'
      AIRFLOW__LOGGING__REMOTE_LOGGING: 'False'
      AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      GOOGLE_APPLICATION_CREDENTIALS: /opt/airflow/secrets/llms-395417-c18ea70a3f54.json
    volumes:
      - ./dags:/opt/airflow/dags
      - ./dbt:/opt/airflow/dbt
      - ./secrets:/opt/airflow/secrets
      - ./scripts:/opt/airflow/scripts
    depends_on:
      - postgres
    command: scheduler

  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres-db-volume:/var/lib/postgresql/data

volumes:
  postgres-db-volume:
docker

1 个回答

  1. Best Answer

    您有两个容器，每个容器都打包了相同的 Dockerfile 和入口点脚本。除了最后运行的命令airflow scheduler或 之外，该脚本完全相同airflow webserver。所有上述命令都在两个容器中运行，包括：

    # Optional: Run DB init & parse DAGs
log "Initializing Airflow DB..."
airflow db upgrade

log "Parsing DAGs..."
airflow scheduler --num-runs 1

DAG_ID="spacex_etl_dag"
log "Unpausing DAG: $DAG_ID"
airflow dags unpause "$DAG_ID" || true

log "Triggering DAG: $DAG_ID"
airflow dags trigger "$DAG_ID" || true

log "Creating admin user (if not exists)..."
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email [email protected] \
    --password admin || true

    尽可能避免在两个镜像中重复执行不必要的命令。可以通过设置多个入口点来实现，或者在入口点处进行检查，以便在相关容器中仅运行必要的步骤。例如，如果只想在调度器容器中运行部分命令（我不熟悉 Airflow，不知道这是否是正确的设计），可以使用以下命令配置脚本：

    log() {
  echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}

if "$1" == "scheduler" ]]; then
  # Optional: Run DB init & parse DAGs
  log "Initializing Airflow DB..."
  airflow db upgrade

  log "Parsing DAGs..."
  airflow scheduler --num-runs 1

  DAG_ID="spacex_etl_dag"
  log "Unpausing DAG: $DAG_ID"
  airflow dags unpause "$DAG_ID" || true

  log "Triggering DAG: $DAG_ID"
  airflow dags trigger "$DAG_ID" || true
fi

log "Creating admin user (if not exists)..."
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email [email protected] \
    --password admin || true

if [[ "$1" == "webserver" || "$1" == "scheduler" ]]; then
  log "Starting Airflow: $1"
  exec airflow "$@"
else
  log "Executing: $@"
  exec "$@"
fi
    • 0

相关问题