在 DAG 中使用可延期运算符

Cloud Composer 1 | Cloud Composer 2

本页面介绍如何在您的环境中启用对可延期运算符的支持，以及如何在您的 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 指标。

除了监控触发器之外，您还可以在环境监控信息中心的未完成任务指标中查看延迟任务的数量。

后续步骤