Airflow 2.0 对 DataprocClusterCreateOperator 的支持答案

【问题标题】：Airflow 2.0 support for DataprocClusterCreateOperatorAirflow 2.0 对 DataprocClusterCreateOperator 的支持
【发布时间】：2022-01-18 15:24:21
【问题描述】：

在我们的项目中，我们使用的是 contrib from airflow.contrib.operators import dataproc_operator 下的 DataprocClusterCreateOperator。气流版本1.10.14 工作正常。

我们正在升级到 Airflow 2.1.2，其中在测试或 dags 需要旋转 DataProc 集群时，我们发现错误为 airflow.exceptions.AirflowException: Invalid arguments were passed to DataprocClusterCreateOperator (task_id: <task_id>). Invalid arguments were: **kwargs: {'config_bucket': None, 'autoscale_policy': None}

我无法在 Airflow 2 中看到此运营商支持的任何链接，因此我可以识别新参数或发生的更改。请分享相关链接。

我们正在使用 google-cloud-composer version 1.17.2 的 Airflow 版本 2.1.2。

【问题讨论】：

Airflow 2.x 中的DataprocClusterCreateOperator 支持的参数可以在源代码中找到here。传递给DataprocClusterCreateOperator 的集群配置参数为here。
另外，正如下面的答案中提到的，可以在here找到运算符的示例实现。

标签： airflow google-cloud-composer airflow-2.x

【解决方案1】：

Airflow 2 中DataprocCreateClusterOperator 支持的参数可以在源代码中找到here。可以传递给算子的集群配置参数可以在here找到。

自 2020 年 1 月 13 日起，DataprocClusterCreateOperator 已根据此 Github commit 重命名为 DataprocCreateClusterOperator，并已从 airflow.contrib.operators 移植到 airflow.providers.google.cloud.operators.dataproc 导入路径。

正如@itroulli 的回答中所给出的，可以在here 找到运算符的示例实现。

【讨论】：

【解决方案2】：

自 Airflow 2.0 起，第 3 方提供商（如本例中的 Google）运营商/挂钩已从 Airflow 核心移至单独的提供商包。你可以阅读更多here。

由于您使用的是 Cloud Composer，因此已经安装了 Google 提供程序包。

关于DataprocClusterCreateOperator，它已重命名为DataprocCreateClusterOperator 并移至airflow.providers.google.cloud.operators.dataproc，因此您可以使用以下命令导入它：

from airflow.providers.google.cloud.operators.dataproc import DataprocCreateClusterOperator

接受的参数不同于 Airflow 1.x 中包含的参数。您可以找到使用示例here。

【讨论】：