在 DAG 中使用可延期运算符

Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3

本页介绍了如何在环境中启用对可延迟操作符的支持，以及如何在 DAG 中使用可延迟 Google Cloud 操作符。

Cloud Composer 中的可推迟运算符简介

如果您有至少一个触发器实例（在高度弹性环境中至少有两个），则可以在 DAG 中使用可延期运算符和触发器。

对于可推迟的运算符，Airflow 会将任务执行分为以下阶段：

  1. 开始操作。在此阶段，任务会占用一个 Airflow 工作器 。该任务会执行一项操作，将作业委托给 不同的服务

    例如，运行一个 BigQuery 作业可能需要 从几秒到几小时创建作业后，该操作会将工作标识符（BigQuery 作业 ID）传递给 Airflow 触发器。

  2. 触发器会监控作业，直到其完成。在此阶段， 未占用工作器槽位。Airflow 触发器具有异步 而且能够处理数百个这样的作业当 触发器检测到作业已完成，就会发送一个事件， 最后一个阶段

  3. 在最后一个阶段，Airflow 工作器执行回调。例如，此回调可以将任务标记为成功，或执行其他操作，并将作业设置为再次由触发器监控。

触发器是无状态的，因此可以应对中断或 重启、.因此，长时间运行的作业对 pod 重启具有弹性，除非重启发生在预计时间较短的最后阶段。

准备工作

  • Cloud Composer 2 中提供了可延期运算符和传感器 并要求满足以下要求：
    • Cloud Composer 2.0.31 及更高版本
    • Airflow 2.2.5、2.3.3 及更高版本

启用对延迟操作符的支持

一个名为 Airflow 触发器的环境组件会异步监控环境中的所有推迟任务。此类任务的延迟操作完成后，触发器会将任务传递给 Airflow 工作器。

您需要在环境中至少有一个触发器实例（在高度弹性环境中至少需要两个），才能在 DAG 中使用可延迟模式。您可以在创建环境时配置触发器，也可以调整现有环境的触发器数量和性能参数。

支持可延期模式的 Google Cloud 运算符

只有部分 Airflow 运算符进行了扩展，以支持可延期模型。 以下列表是对 airflow.providers.google.operators.cloud 软件包中支持延迟模式的运算符的参考。列出所需最低 airflow.providers.google.operators.cloud 软件包版本的列表示该运营商支持延迟模式的最早软件包版本。

Cloud Composer 运算符

运营商名称所需的 apache-airflow-providers-google 版本
CloudComposerCreateEnvironmentOperator 6.4.0
CloudComposerDeleteEnvironmentOperator 6.4.0
CloudComposerUpdateEnvironmentOperator 6.4.0

BigQuery 运算符

运营商名称所需的 apache-airflow-providers-google 版本
BigQueryCheckOperator 8.4.0
BigQueryValueCheckOperator 8.4.0
BigQueryIntervalCheckOperator 8.4.0
BigQueryGetDataOperator 8.4.0
BigQueryInsertJobOperator 8.4.0

BigQuery Data Transfer Service 运算符

运营商名称所需的 apache-airflow-providers-google 版本
BigQueryDataTransferServiceStartTransferRunsOperator 8.9.0

Cloud Build 运算符

运营商名称所需的 apache-airflow-providers-google 版本
CloudBuildCreateBuildOperator 8.7.0

Cloud SQL 运算符

运营商名称所需的 apache-airflow-providers-google 版本
CloudSQLExportInstanceOperator 10.3.0

Dataflow 运算符

运营商名称所需的 apache-airflow-providers-google 版本
DataflowTemplatedJobStartOperator 8.9.0
DataflowStartFlexTemplateOperator 8.9.0

Cloud Data Fusion 运算符

运营商名称所需的 apache-airflow-providers-google 版本
CloudDataFusionStartPipelineOperator 8.9.0

Google Kubernetes Engine 操作器

运营商名称所需的 apache-airflow-providers-google 版本
GKEDeleteClusterOperator 9.0.0
GKECreateClusterOperator 9.0.0

AI Platform 运营商

运营商名称所需的 apache-airflow-providers-google 版本
MLEngineStartTrainingJobOperator 8.9.0

在 DAG 中使用可延期运算符

所有 Google Cloud 运营商的通用惯例是使用 deferrable 布尔值参数启用可延迟模式。如果 Google Cloud 运算符没有此参数，则无法在可推迟模式下运行。其他运算符可能采用不同的惯例。例如，某些社区运营商具有名称中带有 Async 后缀的单独类。

以下示例 DAG 在延迟模式下使用 DataprocSubmitJobOperator 运算符：

PYSPARK_JOB = {
    "reference": { "project_id": "PROJECT_ID" },
    "placement": { "cluster_name": "PYSPARK_CLUSTER_NAME" },
    "pyspark_job": {
        "main_python_file_uri": "gs://dataproc-examples/pyspark/hello-world/hello-world.py"
    },
}

DataprocSubmitJobOperator(
        task_id="dataproc-deferrable-example",
        job=PYSPARK_JOB,
        deferrable=True,
    )

查看触发器日志

触发器生成的日志会与其他 环境组件。如需详细了解如何查看环境日志，请参阅查看日志。

监控触发器

如需详细了解如何监控触发器组件，请参阅 Airflow 指标。

除了监控触发器之外，您还可以查看 您的 Monitoring 信息中心的未完成任务指标中显示的任务 环境

后续步骤