在 Airflow DAG 中使用可延期运算符

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本页面介绍了如何在环境中启用对可延期执行的运算符的支持，以及如何在 DAG 中使用可延期执行的运算符 Google Cloud 。

Cloud Composer 中的可延期执行的运算符简介

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

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

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

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

  2. 触发器会监控作业，直到作业完成。在此阶段，工作线程槽未被占用。Airflow 触发器采用异步架构，能够处理数百个此类作业。当触发器检测到作业已完成时，会发送一个触发最后一个阶段的事件。

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

触发器是无状态的，因此能够应对中断或重启。因此，长时间运行的作业可以应对 pod 重启，除非重启发生在最后一个阶段（预计该阶段很短）。

准备工作

启用对可延迟执行的运算符的支持

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

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

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

只有部分 Airflow 运算符已扩展为支持可延期模型。 以下列表列出了 apache-airflow-providers-google 软件包中支持可延期模式的运算符。 具有最低必需 apache-airflow-providers-google 软件包版本的列表示相应运算符支持可延期模式的最早软件包版本。

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

批处理运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudBatchSubmitJobOperator 10.7.0

Cloud Build 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudBuildCreateBuildOperator 8.7.0

Cloud Composer 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudComposerCreateEnvironmentOperator 6.4.0
CloudComposerDeleteEnvironmentOperator 6.4.0
CloudComposerUpdateEnvironmentOperator 6.4.0
CloudComposerRunAirflowCLICommandOperator 11.0.0

Cloud Run 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudRunExecuteJobOperator 10.7.0

Cloud SQL 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudSQLExportInstanceOperator 10.3.0

Storage Transfer Service 运营商

运算符名称 必需的 apache-airflow-providers-google 版本
CloudDataTransferServiceS3ToGCSOperator 14.0.0
CloudDataTransferServiceGCSToGCSOperator 14.0.0

Dataflow 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
DataflowTemplatedJobStartOperator 8.9.0
DataflowStartFlexTemplateOperator 8.9.0
DataflowStartYamlJobOperator 11.0.0

Cloud Data Fusion 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
CloudDataFusionStartPipelineOperator 8.9.0

Dataplex Universal Catalog 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
DataplexRunDataQualityScanOperator 10.8.0
DataplexGetDataQualityScanResultOperator 10.8.0
DataplexRunDataProfileScanOperator 11.0.0

Google Kubernetes Engine 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
GKEDeleteClusterOperator 9.0.0
GKECreateClusterOperator 9.0.0
GKEStartPodOperator 12.0.0
GKEStartJobOperator 11.0.0

Pub/Sub 运算符

运算符名称 必需的 apache-airflow-providers-google 版本
PubSubPullOperator 14.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 信息中心内查看未完成的任务指标，了解延迟任务的数量。

后续步骤