创建 Managed Service for Apache Airflow 环境

Managed Airflow（第 3 代） | Managed Airflow（第 2 代） | Managed Airflow（旧版第 1 代）

本页面介绍了如何创建 Managed Airflow 环境。

准备工作

第 1 步：创建或选择环境的服务账号

创建环境时，您需要指定服务账号。此服务账号称为“环境的服务账号”。您的环境使用此服务账号执行大部分操作。

您的环境的服务账号不是用户账号。服务账号是由应用或虚拟机 (VM) 实例（而非单个用户）使用的特殊账号。

您以后无法更改环境的服务账号。

如果您的项目中还没有适用于 Managed Airflow 环境的服务账号，请创建该账号。

如需查看有关在 Terraform 中为环境创建服务账号的扩展示例，请参阅创建环境 (Terraform)。

如需为您的环境创建新的服务账号，请执行以下操作：

  1. 按照 Identity and Access Management 文档中的说明创建新的服务账号。

  2. 如 Identity and Access Management 文档中所述，向其授予角色。所需的角色是 Composer Worker (composer.worker)。

  3. 如需访问 Google Cloud 项目中的其他资源，请向此服务账号授予访问这些资源的额外权限。 在大多数情况下，Composer Worker (composer.worker) 角色可提供所需的这组权限。仅当 DAG 的运行需要时，才向此服务账号添加额外权限。

第 2 步：基本设置

此步骤会在指定位置创建一个使用默认参数的 Managed Airflow 环境。

控制台

  1. 在 Google Cloud 控制台中，前往创建环境页面。

    转到“创建环境”

  2. 在名称字段中，为环境输入名称。

    该名称必须以小写字母开头，后面最多可跟 62 个小写字母、数字或连字符，但不能以连字符结尾。该环境名称用于创建环境的子组件，因此您必须提供一个有效的 Cloud Storage 存储桶名称。如需查看限制列表，请参阅存储桶命名准则。

  3. 在位置下拉列表中，为您的环境选择一个位置。

    位置是指环境所在的区域。

  4. 在映像版本下拉列表中，选择所需 Airflow 版本的 Managed Airflow 映像。

  5. 在服务账号下拉列表中，为您的环境选择一个服务账号。

    如果您还没有环境的服务账号，请参阅创建或选择环境的服务账号。

gcloud

gcloud composer environments create ENVIRONMENT_NAME \
    --location LOCATION \
    --image-version IMAGE_VERSION \
    --service-account "SERVICE_ACCOUNT"

将

  • ENVIRONMENT_NAME 替换为环境的名称。

    该名称必须以小写字母开头，后面最多可跟 62 个小写字母、数字或连字符，但不能以连字符结尾。该环境名称用于创建环境的子组件，因此您必须提供一个有效的 Cloud Storage 存储桶名称。如需查看限制列表，请参阅存储桶命名准则。

  • LOCATION 替换为环境的区域。

    位置是指环境所在的区域。

  • 将 SERVICE_ACCOUNT 替换为您的环境的服务账号。

  • 将 IMAGE_VERSION 替换为 Managed Airflow 映像的名称。

示例：

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-3-airflow-2.11.1-build.19 \
    --service-account "
example-account@example-project.
"

API

构建 environments.create API 请求。在 Environment 资源中指定配置。

{
  "name": "projects/PROJECT_ID/locations/LOCATION/environments/ENVIRONMENT_NAME",
  "config": {
    "softwareConfig": {
      "imageVersion": "IMAGE_VERSION"
    },
    "nodeConfig": {
      "serviceAccount": "SERVICE_ACCOUNT"
    }
  }
}

将

  • 将 PROJECT_ID 替换为项目 ID。

  • LOCATION 替换为环境的区域。

    位置是指环境所在的区域。

  • 将 ENVIRONMENT_NAME 替换为环境名称。

    该名称必须以小写字母开头，后面最多可跟 62 个小写字母、数字或连字符，但不能以连字符结尾。该环境名称用于创建环境的子组件，因此您必须提供一个有效的 Cloud Storage 存储桶名称。如需查看限制列表，请参阅存储桶命名准则。

  • 将 IMAGE_VERSION 替换为 Managed Airflow 映像的名称。

  • 将 SERVICE_ACCOUNT 替换为您的环境的服务账号。

例如：

// POST https://composer.googleapis.com/v1/{parent=projects/*/locations/*}/environments

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment",
  "config": {
    "softwareConfig": {
      "imageVersion": "composer-3-airflow-2.11.1-build.19"
    },
    "nodeConfig": {
      "serviceAccount": "
example-account@example-project.
"
    }
  }
}

Terraform

要使用默认参数创建指定位置的环境，请将以下资源块添加到您的 Terraform 配置并运行 terraform apply。

resource "google_composer_environment" "example" {
  provider = google-beta
  name = "ENVIRONMENT_NAME"
  region = "LOCATION"

  config {
    software_config {
      image_version = "IMAGE_VERSION"
    }
    node_config {
      service_account = "SERVICE_ACCOUNT"
    }
  }
}

将

  • ENVIRONMENT_NAME 替换为环境的名称。

    该名称必须以小写字母开头，后面最多可跟 62 个小写字母、数字或连字符，但不能以连字符结尾。该环境名称用于创建环境的子组件，因此您必须提供一个有效的 Cloud Storage 存储桶名称。如需查看限制列表，请参阅存储桶命名准则。

  • LOCATION 替换为环境的区域。

    位置是指环境所在的区域。

  • 将 IMAGE_VERSION 替换为 Managed Airflow 映像的名称。

  • 将 SERVICE_ACCOUNT 替换为您的环境的服务账号。

例如：

resource "google_composer_environment" "example" {
  provider = google-beta
  name = "example-environment"
  region = "us-central1"

  config {
    software_config {
      image_version = "composer-3-airflow-2.11.1-build.19"
    }
    node_config {
      service_account = "
example-account@example-project.
"
    }
  }
}

第 3 步：（可选）配置环境规模和性能参数

如需为您的环境指定扩缩和性能配置，请选择环境大小和工作负载配置。

您可以在创建环境后更改所有性能和扩缩参数。

以下参数可控制规模和性能：

  • 环境大小。控制包括 Airflow 数据库的代管式 Managed Airflow 基础架构的性能参数。如果您希望以更高的基础设施性能运行大量 DAG 和任务，请考虑选择较大的环境大小。例如，环境规模越大，环境可以处理的 Airflow 任务日志条目数量就越多，延迟时间也就越短。

  • 工作负载配置。控制在环境的 GKE 集群中运行的 Airflow 组件的规模和性能。

    • Airflow 调度器。解析 DAG 文件，根据时间表间隔安排 DAG 运行，并将任务排队以待 Airflow 工作器执行。

      您的环境可以同时运行多个 Airflow 调度器。使用多个调度器在多个调度器实例之间分配负载，以实现更好的性能和可靠性。

      增加调度器的数量并不总是可以提高 Airflow 性能。例如，仅使用一个调度器的性能可能优于使用两个调度器。如果额外的调度器未得到利用，则可能会发生这种情况，因而会占用环境的资源，而不会提升整体性能。实际的调度器性能取决于 Airflow 工作器的数量、在您的环境中运行的 DAG 和任务的数量，以及 Airflow 和环境的配置。

      我们建议您先使用两个调度器，然后再监控环境的性能。如果您更改调度器的数量，则可以随时将环境扩缩回原始调度器的数量。

      如需详细了解如何配置多个调度器，请参阅 Airflow 文档。

    • Airflow 触发器。异步监控环境中的所有推迟任务。如果您的环境中至少有一个触发器实例（或在高弹性环境中至少有两个），则可以在 DAG 中使用可延期运算符。

      在 Managed Airflow（第 3 代）中，Airflow 触发器默认处于启用状态。如果您想创建没有触发器的环境，请将触发器的数量设置为零。

    • Airflow DAG 处理器。处理 DAG 文件并将其转换为 DAG 对象。在 Managed Airflow（第 3 代）中，调度程序的这一部分作为单独的环境组件运行。

    • Airflow 网络服务器。运行 Airflow 网页界面，您可在此界面中监控、管理和直观呈现 DAG。

    • Airflow 工作器。执行由 Airflow 调度器执行的任务。环境中的工作器数量下限和上限会动态变化，具体取决于队列中的任务数量。

控制台

您可以为环境选择预设。选择预设时，系统会自动选择该预设的扩缩和性能参数。您还可以选择自定义预设，并指定环境的所有扩缩和性能参数。

如需为您的环境选择容量和性能配置，请在创建环境页面上执行以下操作：

  • 如需使用预定义的值，请在环境资源部分中点击小、中、大或超大。

  • 要为规模和性能参数指定自定义值，请执行以下操作：

    1. 在环境资源部分中，点击自定义。

    2. 在调度器部分中，设置要使用的调度器数量，以及为调度器分配的 CPU、内存和存储空间资源。

    3. 在 Triggerer 部分中，使用触发器数量字段输入环境中的触发器数量。

      如果您不想在 DAG 中使用可延期运算符，请将触发器的数量设置为零。

      如果您为环境设置了至少一个触发器，请使用 CPU 和 内存字段为触发器配置资源分配。

    4. 在 DAG 处理器部分中，指定环境中的 DAG 处理器数量以及每个 DAG 处理器的 CPU、内存和存储空间量。

      高弹性环境至少需要两个 DAG 处理器。

    5. 在Web 服务器部分中，指定 Web 服务器的 CPU、内存和存储空间量。

    6. 在工作器部分，指定：

      • 您环境中的自动扩缩工作器数量下限和上限。
      • 工作器的 CPU、内存和存储空间分配

    7. 在核心基础架构部分的环境大小下拉列表中，选择环境大小。

gcloud

创建环境时，您可以通过以下参数控制环境的规模和性能参数。

  • --environment-size 用于指定环境大小。
  • --scheduler-count 用于指定调度器的数量。
  • --scheduler-cpu 用于指定 Airflow 调度器的 CPU 数量。
  • --scheduler-memory 用于指定 Airflow 调度器的内存量。

  • --scheduler-storage 用于指定 Airflow 调度器的磁盘空间。

  • --triggerer-count 用于指定环境中的 Airflow 触发器数量。此标志的默认值为 0。 如果您想在 DAG 中使用可延期运算符，则需要触发器。

    • 对于标准弹性环境，请使用介于 0 和 10 之间的值。
    • 对于高弹性环境，请使用 0 或介于 2 和 10 之间的值。

  • --triggerer-cpu 用于指定 Airflow 触发器的 CPU 数量（以 vCPU 为单位）。允许使用的值有：0.5、0.75、1。默认值为 0.5。

  • --triggerer-memory 用于指定 Airflow 触发器的内存量（以 GB 为单位）。默认值为 0.5。

    所需的最低内存等于为触发器分配的 CPU 数量。允许的最大值等于触发器 CPU 的数量乘以 6.5。

    例如，如果您将 --triggerer-cpu 标志设置为 1，则 --triggerer-memory 的最小值为 1，最大值为 6.5。

  • --dag-processor-count 用于指定环境中的 DAG 处理器数量。

    高弹性环境至少需要两个 DAG 处理器。

  • --dag-processor-cpu 用于指定 DAG 处理器的 CPU 数量。

  • --dag-processor-memory 用于指定 DAG 处理器的内存量。

  • --dag-processor-storage 用于指定 DAG 处理器的磁盘空间量。

  • --web-server-cpu 用于指定 Airflow Web 服务器的 CPU 数量。

  • --web-server-memory 用于指定 Airflow Web 服务器的内存量。

  • --web-server-storage 用于指定 Airflow Web 服务器的磁盘空间。

  • --worker-cpu 用于指定 Airflow 工作器的 CPU 数量。

  • --worker-memory 用于指定 Airflow 工作器的内存量。

  • --worker-storage 用于指定 Airflow 工作器的磁盘空间。

  • --min-workers 用于指定 Airflow 工作器数量下限。您的环境的集群至少运行此数量的工作器。

  • --max-workers 用于指定 Airflow 工作器数量上限。您的环境的集群最多运行此数量的工作器。

gcloud composer environments create ENVIRONMENT_NAME \
    --location LOCATION \
    --image-version composer-3-airflow-2.11.1-build.19 \
    --service-account "SERVICE_ACCOUNT" \
    --environment-size ENVIRONMENT_SIZE \
    --scheduler-count SCHEDULER_COUNT \
    --scheduler-cpu SCHEDULER_CPU \
    --scheduler-memory SCHEDULER_MEMORY \
    --scheduler-storage SCHEDULER_STORAGE \
    --triggerer-count TRIGGERER_COUNT \
    --triggerer-cpu TRIGGERER_CPU \
    --triggerer-memory TRIGGERER_MEMORY \
    --dag-processor-count DAG_PROCESSOR_COUNT \
    --dag-processor-cpu DAG_PROCESSOR_CPU \
    --dag-processor-memory DAG_PROCESSOR_MEMORY \
    --dag-processor-storage DAG_PROCESSOR_STORAGE \
    --web-server-cpu WEB_SERVER_CPU \
    --web-server-memory WEB_SERVER_MEMORY \
    --web-server-storage WEB_SERVER_STORAGE \
    --worker-cpu WORKER_CPU \
    --worker-memory WORKER_MEMORY \
    --worker-storage WORKER_STORAGE \
    --min-workers WORKERS_MIN \
    --max-workers WORKERS_MAX

将

  • ENVIRONMENT_SIZE，包含 small、medium、large、extra-large。
  • SCHEDULER_COUNT 替换为调度器的数量。
  • 将 SCHEDULER_CPU 替换为调度器的 CPU 数量（以 vCPU 为单位）。
  • 将 SCHEDULER_MEMORY 替换为调度器的内存量。
  • 将 SCHEDULER_STORAGE 替换为调度器的磁盘大小。
  • 将 TRIGGERER_COUNT 替换为触发器的数量。
  • 将 TRIGGERER_CPU 替换为触发器的 CPU 数量（以 vCPU 为单位）。

  • 将 TRIGGERER_MEMORY 替换为触发器的内存量（以 GB 为单位）。

  • 将 DAG_PROCESSOR_COUNT 替换为 DAG 处理器数量。

  • 将 DAG_PROCESSOR_CPU 替换为 DAG 处理器的 CPU 数量。

  • 将 DAG_PROCESSOR_MEMORY 替换为 DAG 处理器的内存量。

  • 将 DAG_PROCESSOR_STORAGE 替换为 DAG 处理器的磁盘空间量。

  • 将 WEB_SERVER_CPU 替换为网络服务器的 CPU 数量（以 vCPU 为单位）。

  • 将 WEB_SERVER_MEMORY 替换为网络服务器的内存量。

  • 将 WEB_SERVER_STORAGE 替换为网络服务器的内存量。

  • 将 WORKER_CPU 替换为工作器的 CPU 数量（以 vCPU 为单位）。

  • 将 WORKER_MEMORY 替换为工作器的内存量。

  • 将 WORKER_STORAGE 替换为工作器的磁盘大小。

  • 将 WORKERS_MIN 替换为您的环境可运行的 Airflow 工作器数量的下限。您的环境中的工作器数量不能低于此数量，即使更少数量的工作器可以处理负载也是如此。

  • 将 WORKERS_MAX 替换为您的环境可运行的 Airflow 工作器数量的上限。您的环境中的工作器数量不能超出此数量，即使需要更多数量的工作器来处理负载也是如此。

例如：

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-3-airflow-2.11.1-build.19 \
    --service-account "
example-account@example-project.
" \
    --environment-size small \
    --scheduler-count 1 \
    --scheduler-cpu 0.5 \
    --scheduler-memory 2.5GB \
    --scheduler-storage 2GB \
    --triggerer-count 1 \
    --triggerer-cpu 0.5 \
    --triggerer-memory 0.5GB \
    --dag-processor-count 1 \
    --dag-processor-cpu 0.5 \
    --dag-processor-memory 2GB \
    --dag-processor-storage 1GB \
    --web-server-cpu 1 \
    --web-server-memory 2.5GB \
    --web-server-storage 2GB \
    --worker-cpu 1 \
    --worker-memory 2GB \
    --worker-storage 2GB \
    --min-workers 2 \
    --max-workers 4

API

创建环境时，请在环境 > EnvironmentConfig > WorkloadsConfig 资源中指定环境规模和性能参数。

{
  "name": "projects/PROJECT_ID/locations/LOCATION/environments/ENVIRONMENT_NAME",
  "config": {
    "workloadsConfig": {
      "scheduler": {
        "cpu": SCHEDULER_CPU,
        "memoryGb": SCHEDULER_MEMORY,
        "storageGb": SCHEDULER_STORAGE,
        "count": SCHEDULER_COUNT
      },
      "triggerer": {
        "count": TRIGGERER_COUNT,
        "cpu": TRIGGERER_CPU,
        "memoryGb": TRIGGERER_MEMORY
      },
      "dagProcessor": {
        "count": DAG_PROCESSOR_COUNT,
        "cpu": DAG_PROCESSOR_CPU,
        "memoryGb": DAG_PROCESSOR_MEMORY,
        "storageGb": DAG_PROCESSOR_STORAGE
      },
      "webServer": {
        "cpu": WEB_SERVER_CPU,
        "memoryGb": WEB_SERVER_MEMORY,
        "storageGb": WEB_SERVER_STORAGE
      },
      "worker": {
        "cpu": WORKER_CPU,
        "memoryGb": WORKER_MEMORY,
        "storageGb": WORKER_STORAGE,
        "minCount": WORKERS_MIN,
        "maxCount": WORKERS_MAX
      }
    },
    "environmentSize": "ENVIRONMENT_SIZE",
    "nodeConfig": {
      "serviceAccount": "SERVICE_ACCOUNT"
    }
  }
}

替换：

  • 将 SCHEDULER_CPU 替换为调度器的 CPU 数量（以 vCPU 为单位）。
  • 将 SCHEDULER_MEMORY 替换为调度器的内存量（以 GB 为单位）。
  • 将 SCHEDULER_STORAGE 替换为调度器的磁盘大小（以 GB 为单位）。

  • SCHEDULER_COUNT 替换为调度器的数量。

  • TRIGGERER_COUNT 替换为触发器的数量。默认值为 0。 如果您想在 DAG 中使用可延期运算符，则需要触发器。

    • 对于标准弹性环境，请使用介于 0 和 10 之间的值。
    • 对于高弹性环境，请使用 0 或介于 2 和 10 之间的值。

    如果您使用至少一个触发器，还必须指定 TRIGGERER_CPU 和 TRIGGERER_MEMORY 参数：

  • TRIGGERER_CPU 用于指定触发器的 CPU 数量（以 vCPU 为单位）。允许使用的值有：0.5、0.75、1。

  • TRIGGERER_MEMORY 用于配置触发器的内存量。所需内存的最小值等于为触发器分配的 CPU 数量。允许的最大值等于触发器 CPU 的数量乘以 6.5。

    例如，如果您将 TRIGGERER_CPU 设置为 1，则 TRIGGERER_MEMORY 的最小值为 1，最大值为 6.5。

  • 将 DAG_PROCESSOR_COUNT 替换为 DAG 处理器数量。

    高弹性环境至少需要两个 DAG 处理器。

  • 将 DAG_PROCESSOR_CPU 替换为 DAG 处理器的 CPU 数量（以 vCPU 为单位）。

  • 将 DAG_PROCESSOR_MEMORY 替换为 DAG 处理器的内存量（以 GB 为单位）。

  • 将 DAG_PROCESSOR_STORAGE 替换为 DAG 处理器的磁盘空间量（以 GB 为单位）。

  • 将 WEB_SERVER_CPU 替换为网络服务器的 CPU 数量（以 vCPU 为单位）。

  • 将 WEB_SERVER_MEMORY 替换为 Web 服务器的内存量（以 GB 为单位）。

  • 将 WEB_SERVER_STORAGE 替换为 Web 服务器的磁盘大小（以 GB 为单位）。

  • 将 WORKER_CPU 替换为工作器的 CPU 数量（以 vCPU 为单位）。

  • 将 WORKER_MEMORY 替换为工作器的内存量（以 GB 为单位）。

  • 将 WORKER_STORAGE 替换为工作器的磁盘大小（以 GB 为单位）。

  • 将 WORKERS_MIN 替换为您的环境可运行的 Airflow 工作器数量的下限。您的环境中的工作器数量不能低于此数量，即使更少数量的工作器可以处理负载也是如此。

  • 将 WORKERS_MAX 替换为您的环境可运行的 Airflow 工作器数量的上限。您的环境中的工作器数量不能超出此数量，即使需要更多数量的工作器来处理负载也是如此。

  • 将 ENVIRONMENT_SIZE 替换为环境大小：ENVIRONMENT_SIZE_SMALL、ENVIRONMENT_SIZE_MEDIUM、ENVIRONMENT_SIZE_LARGE、ENVIRONMENT_SIZE_EXTRA_LARGE。

示例：

// POST https://composer.googleapis.com/v1/{parent=projects/*/locations/*}/environments

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment",
  "config": {
    "workloadsConfig": {
      "scheduler": {
        "cpu": 2.5,
        "memoryGb": 2.5,
        "storageGb": 2,
        "count": 1