在 Monitoring 信息中心内使用关键指标监控环境健康状况和性能

Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3

本页面介绍了如何监控整个 Cloud Composer 监控环境健康状况和性能，监控环境健康状况和性能。

简介

本教程重点介绍关键的 Cloud Composer 监控指标 可以很好地概要了解环境级运行状况和性能。

Cloud Composer 提供多种指标， 环境状态。本教程中的监控准则基于 （针对 Monitoring 信息中心上显示的指标） 使用 Cloud Composer

在本教程中，您将了解 环境性能和健康状况问题的主要指标， 以及将每项指标解读为纠正措施的指南， 保持环境健康您还需要为每项服务 运行示例 DAG 并使用这些指标和提醒来优化 环境性能

目标

目标：

• 了解 Monitoring 上提供的关键性能和健康指标 信息中心。
• 了解每个指标的基准。
• 了解如何将指标解读为纠正措施。
• 为每个关键指标创建提醒。
• 运行示例 DAG。
• 将提醒和 Monitoring 图表解读为纠正措施。
• 优化您的环境并评估其性能。

费用

本教程使用 Google Cloud 的以下收费组件：

• Cloud Composer（请参阅 额外费用）
• Cloud Monitoring

完成本教程后，您可以删除所创建的资源以避免继续计费。如需了解详情，请参阅清理。

准备工作

本部分介绍了在开始学习本教程之前必须执行的操作。

创建和配置项目

在本教程中，您需要 project。 按以下方式配置项目：

1. 在 Google Cloud 控制台中，选择或创建项目：

   转到项目选择器

2. 确保您的项目已启用结算功能。 了解如何检查项目是否已启用结算功能。

3. 如需创建必要的资源，请确保您的 Google Cloud 项目用户具有以下角色：

   • 环境和存储对象管理员 (roles/composer.environmentAndStorageObjectAdmin)
   • Compute Admin (roles/compute.admin)
   • Monitoring Editor (roles/monitoring.editor)

为您的项目启用 API

启用 Cloud Composer API。

启用 API

创建 Cloud Composer 环境

创建一个 Cloud Composer 2 环境。

在此过程中 您授予 Cloud Composer v2 API Service Agent Extension (roles/composer.ServiceAgentV2Ext) 角色分配给 Composer Service Agent 。Cloud Composer 使用此账号执行操作 Google Cloud 项目中的资源。

探索有关环境级健康状况和性能的关键指标

本教程重点介绍了可帮助您全面了解 环境的整体健康状况和性能。

Monitoring 信息中心 Google Cloud 控制台包含各种指标和图表 监控环境中的趋势并发现 Airflow 问题 组件和 Cloud Composer 资源

每个 Cloud Composer 环境都有自己的 Monitoring 信息中心。

熟悉以下关键指标，并在 Monitoring 信息中心内找到每个指标：

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

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 转到监控标签页。

4. 选择概览部分，然后在以下位置找到环境概览项 查看信息中心 环境运行状况（Airflow 监控 DAG）指标。

   • 此时间轴显示了 Cloud Composer 的健康状况 环境环境运行状况栏的绿色 表示环境健康状况良好，而环境状况不佳则用红色表示。

   • 每隔几分钟，Cloud Composer 就会执行一个名为 airflow_monitoring。如果活跃性 DAG 运行成功完成， 健康状况为True。如果活跃性 DAG 运行失败（例如， （由于 Pod 逐出、外部进程终止或维护）， 健康状况为 False。

5. 选择 SQL 数据库部分，找到数据库运行状况 项目，并观察数据库运行状况指标。

   • 此时间轴显示了 您的环境的 Cloud SQL 实例。绿色数据库 运行状况栏表示连接正常，而连接失败则表示 以红色表示

   • Airflow 监控 Pod 会定期 ping 数据库并报告 如果可以建立连接，则为 True 或 如果不是，则为 False。

6. 在 Database health 项中，观察数据库 CPU 使用率和 数据库内存用量指标。

   • 数据库 CPU 使用率图表通过 您环境的 Cloud SQL 数据库实例与 可用数据库的总 CPU 限制。

   • “数据库内存使用量”图表显示 您环境的 Cloud SQL 数据库实例与 可用数据库内存的总上限。

7. 选择调度器部分，找到调度器检测信号 项，并观察调度器检测信号指标。

   • 此时间轴显示了 Airflow 调度器的健康状况。 检查红色区域，以识别 Airflow 调度器问题。如果您的 环境有多个调度器，则检测信号状态为 运行状况良好，前提是至少有一个调度器正在响应。

   • 如果收到最后一次检测信号，则调度程序会被视为健康状况不佳 当前时间之前超过 30 秒（默认值）。

8. 选择 DAG 统计信息部分，找到已终止的僵尸任务 项目，并查看 Zombie tasks killed 指标。

   • 这张图显示了小程序中终止的僵尸任务数量 。僵尸任务通常是由外部终止 Airflow 进程的影响（例如，任务进程被终止时）。

   • Airflow 调度器会定期终止僵尸任务，这也会 数据。

9. 选择 Workers 部分，找到 Worker container restarts。 项目，并观察工作器容器重启次数指标。

   • 显示单个工作器的重启总次数的图表 容器容器重启次数过多可能会影响 将其作为依赖项的服务或其他下游服务。

。

注意： 您可以通过 在 Cloud Monitoring 中查看。

了解关键指标的基准和可能的纠正措施

重要提示： 时间轴指标可让您大致了解 各个环境组件；不过，红色区域并不一定是 表明某个组件健康状况不佳。有时 整体绿色时间轴可能是由外部因素造成的，例如维护 环境的集群中。使用相应的 Monitoring 图表和日志来调查潜在原因，分析相关性 并根据已确定的根本原因进行更正 原因。

以下列表介绍了基准值，这些基准值可指明 提供了可用于解决这些问题的纠正措施。

• 环境健康状况（Airflow 监控 DAG）

  • 在 4 小时的时间段内，成功率低于 90%

  • 失败可能意味着 Pod 被逐出或工作器终止，因为 或发生故障环境中的红色区域 健康状况时间轴通常与其他健康状况条中的红色区域相关 各个环境组件的影响。找出根本原因 在 Monitoring 信息中心内查看其他指标。

• 数据库运行状况

  • 在 4 小时的时间段内，成功率低于 95%

  • 失败表示与 Airflow 连接出现问题 这可能是由于数据库崩溃或停机引起， 数据库过载（例如，由于 CPU 或内存 或连接数据库期间的延迟较高）。这些症状 最常见的原因是不够最佳的 DAG，例如当 DAG 使用 多个全局定义的 Airflow 或环境变量。确定根 请查看 SQL 数据库资源使用情况指标。您还可以检查调度器日志中是否存在与数据库连接相关的错误。

• 数据库 CPU 和内存用量

  • 在 12 小时的时间段内，CPU 或内存的平均使用率超过 80%

  • 数据库可能已过载。分析 DAG 之间的相关性 运行和出现数据库 CPU 或内存用量高峰。

    • 您可以通过以下方式减少数据库负载： 更加高效的 DAG 优化型运行的查询和连接，或者通过将 会随着时间的推移而更均匀地加载。

    • 或者，您也可以为数据库分配更多 CPU 或内存。 数据库资源由 您的环境，并且该环境必须是 放大为更大的尺寸。

• 调度程序检测信号

  • 在 4 小时的时间段内，成功率低于 90%

  • 为调度器分配更多资源或增加调度器数量 从 1 到 2（推荐）。

• 已终止的僵尸任务

  • 每 24 小时内只能执行一项僵尸任务

  • 执行僵尸任务的最常见原因是 CPU 或内存不足 集群内的资源查看工作器资源使用情况 并为工作器分配更多资源，或者 延长僵尸任务的超时时长 这样调度程序在将任务视为僵尸前等待更长时间。

• 工作器容器重启

  • 每 24 小时重启多次

  • 最常见的原因是缺少工作器内存或存储空间。调查 消耗工作器资源，以及分配更多内存或存储空间 。如果原因不是资源不足，请调查 排查工作器重启突发事件 并使用 日志记录查询 来找出工作器重启的原因。

创建通知渠道

请按照 创建通知渠道以创建电子邮件通知渠道。

注意 ：为实现冗余，建议您创建多个类型 通知渠道。例如，您可以设置一个 Slack 通知， 电子邮件渠道。

如需详细了解通知渠道，请参阅 管理通知渠道。

创建提醒政策

根据上一部分中提供的基准创建提醒政策 持续监控指标的值 在这些指标违反条件时收到通知。

控制台

您可以通过以下方式为 Monitoring 信息中心中显示的每个指标设置提醒： 点击相应项一角的铃铛图标：

<ph type="x-smartling-placeholder"></ph> <ph type="x-smartling-placeholder">

</ph> 图 1.为监控信息中心上显示的指标创建提醒（点击可放大）

。

1. 在 Monitoring 中找到要监控的每个指标 信息中心，然后点击指标项一角的铃铛图标。通过 创建提醒政策页面即会打开。

   重要提示 ：指标名称、环境名称和位置过滤条件 但系统自动应用的某些提醒政策， 指标 需要额外的过滤条件和不同的指标名称。 请参阅“过滤条件”字段和“指标名称”在提醒政策中 配置一个单独的指标，以验证正确的过滤器 应用。

2. 在转换数据部分中，执行以下操作：

   1. 按照 指标的提醒政策配置。

   2. 点击下一步，然后配置配置提醒触发器部分 如指标的提醒政策配置中所述。

3. 点击下一步。

4. 配置通知。展开通知渠道菜单 并选择您在 上一步。

5. 点击 OK（确定）。

6. 在为提醒政策命名部分，填写提醒政策名称。 字段。为每个指标使用描述性名称。使用“命名 提醒政策”值，如提醒政策配置 指标的值

7. 点击下一步。

8. 查看提醒政策，然后点击创建政策。

环境健康状况（Airflow 监控 DAG）指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 健康状况良好
• API：composer.googleapis.com/environment/healthy

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：true 分数

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：低于阈值
  • 阈值：90
  • 条件名称：环境健康状况

• 配置通知并完成提醒：

  • 将提醒政策命名为“Airflow Environment Health”

数据库运行状况指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 数据库健康状况
• API：composer.googleapis.com/environment/database_health

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：true 分数

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：低于阈值
  • 阈值：95
  • 条件名称：数据库健康状况

• 配置通知并完成提醒：

  • 将提醒政策命名为：Airflow Database Health

数据库 CPU 用量指标 - 提醒政策配置

注意 ：为简单起见，在设置数据库内存用量提醒和 数据库 CPU 使用率指标，请使用指标名称 数据库内存利用率 (composer.googleapis.com/environment/database/memory/utilization) 和数据库 CPU 利用率 (composer.googleapis.com/environment/database/cpu/utilization)，而不是 监控已使用字节数的默认指标名称 （composer.googleapis.com/environment/database/memory/bytes_used）和二手 时间 (composer.googleapis.com/environment/database/cpu/usage_time)。通过 数据库内存和 CPU 利用率指标用于监控 这有助于避免不必要的 计算。

• 指标名称：Cloud Composer 环境 - 数据库 CPU 利用率
• API：composer.googleapis.com/environment/database/cpu/utilization

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：自定义
  • 自定义值：12
  • 自定义单位：小时
  • 滚动窗口函数：均值

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：高于阈值
  • 阈值：80
  • 条件名称：数据库 CPU 使用条件

• 配置通知并完成提醒：

  • 将提醒政策命名为：Airflow Database CPU Usage

数据库 CPU 用量指标 - 提醒政策配置

注意 ：为简单起见，在设置数据库内存用量提醒和 数据库 CPU 使用率指标，请使用指标名称 数据库内存利用率 (composer.googleapis.com/environment/database/memory/utilization) 和数据库 CPU 利用率 (composer.googleapis.com/environment/database/cpu/utilization)，而不是 监控已使用字节数的默认指标名称 （composer.googleapis.com/environment/database/memory/bytes_used）和二手 时间 (composer.googleapis.com/environment/database/cpu/usage_time)。通过 数据库内存和 CPU 利用率指标用于监控 这有助于避免不必要的 计算。

• 指标名称：Cloud Composer 环境 - 数据库内存利用率
• API：composer.googleapis.com/environment/database/memory/utilization

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：自定义
  • 自定义值：12
  • 自定义单位：小时
  • 滚动窗口函数：均值

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：高于阈值
  • 阈值：80
  • 条件名称：数据库内存用量

• 配置通知并完成提醒：

  • 将提醒政策命名为：Airflow Database Memory Usage

调度器检测信号指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 调度器检测信号
• API：composer.googleapis.com/environment/scheduler_heartbeat_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：count

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：低于阈值

  • 阈值：216

    注意 ：空闲环境中的平均调度器检测信号计数为 240，因此 4 小时的成功率为 90%，等于 240 * 0.9 =
    1. 您可以通过运行汇总值的查询来获取该数字。 _scheduler_heartbeat_count_mean在 Metrics Explorer 查询编辑器。

  • 条件名称：调度器检测信号条件

• 配置通知并完成提醒：

  • 将提醒政策命名为：Airflow 调度程序检测信号

“已终止的僵尸任务数”指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 已终止的僵尸任务
• API：composer.googleapis.com/environment/zombie_task_killed_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：1 天
  • 滚动窗口函数：sum

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：高于阈值
  • 阈值：1
  • 条件名称：僵尸任务条件

• 配置通知并完成提醒：

  • 将提醒政策命名为“Airflow Zombie Tasks”。

工作器容器重启指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 已终止的僵尸任务
• API：composer.googleapis.com/environment/zombie_task_killed_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 &gt; 在每个时序内：

  • 滚动窗口：1 天
  • 滚动窗口函数：sum

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何违反时序的情况
  • 阈值位置：高于阈值
  • 阈值：1
  • 条件名称：僵尸任务条件

• 配置通知并完成提醒：

  • 将提醒政策命名为“Airflow Zombie Tasks”。

Terraform

运行一个用于创建电子邮件通知渠道并上传内容的 Terraform 脚本 为本教程中提供的关键指标制定提醒政策， 各自的基准：

1. 将示例 Terraform 文件保存在本地计算机上。

2. 替换以下内容：

   • PROJECT_ID：项目 ID 项目名称例如 example-project。
   • EMAIL_ADDRESS：电子邮件地址，如果 提醒。
   • ENVIRONMENT_NAME：您的 Cloud Composer 环境的名称。 例如 example-composer-environment。
   • CLUSTER_NAME：您的环境集群名称，可以在以下位置找到： 环境配置 &gt; 资源 &gt; GKE Google Cloud 控制台中的集群。

resource "google_monitoring_notification_channel" "basic" {
  project      = "PROJECT_ID"
  display_name = "Test Notification Channel"
  type         = "email"
  labels = {
    email_address = "EMAIL_ADDRESS"
  }
  # force_delete = false
}

resource "google_monitoring_alert_policy" "environment_health_metric" {
  project      = "PROJECT_ID"
  display_name = "Airflow Environment Health"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Environment health condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/healthy\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 0.9
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_FRACTION_TRUE"
      }
    }
  }

}

resource "google_monitoring_alert_policy" "database_health_metric" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database Health"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database health condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database_health\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 0.95
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_FRACTION_TRUE"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_database_cpu_usage" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database CPU Usage"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database CPU usage condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database/cpu/utilization\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 80
      aggregations {
        alignment_period   = "43200s"
        per_series_aligner = "ALIGN_MEAN"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_database_memory_usage" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database Memory Usage"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database memory usage condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database/memory/utilization\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 80
      aggregations {
        alignment_period   = "43200s"
        per_series_aligner = "ALIGN_MEAN"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_scheduler_heartbeat" {
  project      = "PROJECT_ID"
  display_name = "Airflow Scheduler Heartbeat"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Scheduler heartbeat condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/scheduler_heartbeat_count\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 216 // Threshold is 90% of the average for composer.googleapis.com/environment/scheduler_heartbeat_count metric in an idle environment
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_COUNT"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_zombie_task" {
  project      = "PROJECT_ID"
  display_name = "Airflow Zombie Tasks"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Zombie tasks condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/zombie_task_killed_count\" AND  resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 1
      aggregations {
        alignment_period   = "86400s"
        per_series_aligner = "ALIGN_SUM"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_worker_restarts" {
  project      = "PROJECT_ID"
  display_name = "Airflow Worker Restarts"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Worker container restarts condition"
    condition_threshold {
      filter     = "resource.type = \"k8s_container\" AND (resource.labels.cluster_name = \"CLUSTER_NAME\" AND resource.labels.container_name = monitoring.regex.full_match(\"airflow-worker|base\") AND resource.labels.pod_name = monitoring.regex.full_match(\"airflow-worker-.*|airflow-k8s-worker-.*\")) AND metric.type = \"kubernetes.io/container/restart_count\""

      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 1
      aggregations {
        alignment_period   = "86400s"
        per_series_aligner = "ALIGN_RATE"
      }
    }
  }
}

测试提醒政策

本部分介绍如何测试创建的提醒政策并解读 结果。

上传示例 DAG

本教程中提供的示例 DAG memory_consumption_dag.py 模拟了 需要较高的工作器内存利用率该 DAG 包含 4 项任务，每项任务 任务将数据写入示例字符串，消耗 380 MB 的内存。示例 DAG 安排为每 2 分钟运行一次，将自动开始运行 然后再上传到 Composer 环境中。

警告 ：学完本教程后，请务必删除或暂停此 DAG。 否则您的环境会消耗大量的资源 可能会产生额外的费用。

将以下示例 DAG 上传到环境中 您在之前步骤中创建的应用：

from datetime import datetime
import sys
import time

from airflow import DAG
from airflow.operators.python import PythonOperator


def ram_function():
    data = ""
    start = time.time()
    for i in range(38):
        data += "a" * 10 * 1000**2
        time.sleep(0.2)
        print(f"{i}, {round(time.time() - start, 4)}, {sys.getsizeof(data) / (1000 ** 3)}")
    print(f"Size={sys.getsizeof(data) / (1000 ** 3)}GB")
    time.sleep(30 - (time.time() - start))
    print(f"Complete in {round(time.time() - start, 2)} seconds!")


with DAG(
    dag_id="memory_consumption_dag",
    start_date=datetime(2023, 1, 1, 1, 1, 1),
    schedule="1/2 * * * *",
    catchup=False,
) as dag:
    for i in range(4):
        PythonOperator(
            task_id=f"task_{i+1}",
            python_callable=ram_function,
            retries=0,
            dag=dag,
        )

解读 Monitoring 中的提醒和指标

在示例 DAG 开始运行后等待大约 10 分钟，然后评估 测试结果：

1. 查看您的电子邮件邮箱，以确认是否收到了来自 主题行以 [ALERT] 开头的 Google 云提醒。 此消息的内容包含提醒政策突发事件详细信息。

2. 点击电子邮件通知中的查看突发事件按钮。您 已重定向到 Metrics Explorer。查看提醒的详细信息 突发事件：

   <ph type="x-smartling-placeholder"></ph> <ph type="x-smartling-placeholder">

   </ph> 图 2.提醒突发事件的详细信息（点击可放大）

   。

   突发事件指标图表会指出您创建的指标超出了 阈值为 1，表示 Airflow 检测到并击败了超过 1 只僵尸 任务。

3. 在 Cloud Composer 环境中，转到监控标签页。 打开 DAG 统计信息部分，然后找到已终止的僵尸任务 图表：

   <ph type="x-smartling-placeholder"></ph> <ph type="x-smartling-placeholder">

   </ph> 图 3.僵尸任务图表（点击可放大）

   。

   该图显示，Airflow 在 运行示例 DAG 的前 10 分钟。

4. 根据基准和纠正措施，最常见的原因 缺乏工作器内存或 CPU。确定根本原因 通过分析工作器资源利用率来分析僵尸任务。

   在 Monitoring 信息中心内打开“工作器”部分，并查看工作器 CPU 和内存使用情况指标：

   <ph type="x-smartling-placeholder"></ph> <ph type="x-smartling-placeholder">

   </ph> 图 4. 工作器 CPU 和内存使用情况指标（点击可放大）

   工作器 CPU 使用率图表显示，工作器 CPU 使用率 始终低于可用总上限的 50%，因此 就足够了工作器内存用量图表显示，运行 示例 DAG 导致达到可分配内存上限，该上限等于 图表上显示的总内存限额的近 75% （GKE 会预留前 4 GiB 的 25% 的内存， 在每个节点上额外增加 100 MiB 内存来处理 Pod 逐出）。

   由此可以得出结论，工作器缺少运行 已成功创建 DAG 示例。

优化您的环境并评估其性能

根据对工作器资源利用率的分析，您需要分配更多 向工作器分配内存，确保 DAG 中的所有任务都成功完成。

注意 ：更改 worker 的CPU 或内存重启工作器，这可能会影响 运行任务。我们建议您在没有 DAG 运行时执行此操作。

1. 在 Composer 环境中，打开 DAG 标签页，然后点击 示例 DAG (memory_consumption_dag)，然后点击暂停 DAG。

2. 分配额外的工作器内存：

   1. 在“环境配置”标签页中，找到资源 &gt; 工作负载配置，然后点击修改。

   2. 在 Worker 项中，提高内存限制。在本教程中 使用 3.25 GB。

   3. 保存更改，然后等待几分钟，让工作器重启。

3. 打开 DAG 标签页，点击示例 DAG 的名称 (memory_consumption_dag)，然后点击取消暂停 DAG。

转到 Monitoring，确认在您之后没有显示新的僵尸任务 更新了工作器资源限制：

<ph type="x-smartling-placeholder"></ph> <ph type="x-smartling-placeholder">

</ph> 图 5.内存限制发生变化后的僵尸任务图表（点击可放大）

。

注意： 建议观察在延长期间的环境性能 。例如，在您运行多个 DAG 的生产环境中， 建议您在 24 小时后仔细检查是否没有新的提醒。

摘要

在本教程中，您了解了环境层面的关键运行状况和 性能指标，如何为每个指标设置提醒政策，以及如何 将每项指标解读为纠正措施。然后，您运行了一个示例 DAG 借助警报确定环境健康问题的根本原因 和 Monitoring 图表，并通过分配更多内存来优化您的环境 。不过，建议您 优化 DAG 减少工作器资源消耗量 将资源增加到超过特定阈值。

清理

为避免系统因资源向您的 Google Cloud 账号收取费用 您可以删除包含这些资源的项目 或者保留项目而删除各个资源。

删除项目

警告：删除项目会产生以下影响

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目可以帮助您避免超出项目配额限制。

1. 在 Google Cloud 控制台中，进入管理资源页面。

   转到“管理资源”

2. 在项目列表中，选择要删除的项目，然后点击删除。
3. 在对话框中输入项目 ID，然后点击关闭以删除项目。

删除各个资源

如果您打算探索多个教程和快速入门，重复使用项目可以帮助您避免超出项目配额上限。

控制台

1. 删除 Cloud Composer 环境。您 删除环境的存储桶
2. 删除每项提醒政策 您在 Cloud Monitoring 中创建的应用。

Terraform

1. 确保您的 Terraform 脚本不包含 项目仍需要的资源例如，您 可能希望让一些 API 保持启用状态， （如果您已将此类定义添加到 Terraform 脚本）。
2. 运行 terraform destroy。
3. 手动删除环境的存储桶。Cloud Composer 不会自动将其删除您可以从 Google Cloud 控制台或 Google Cloud CLI。

后续步骤

• 优化环境
• 扩缩环境
• 管理环境标签并细分环境费用