Google Cloud 连接¶

Google Cloud 连接类型支持 Google Cloud 集成。

认证到 Google Cloud¶

使用 Airflow 连接 Google Cloud 有三种方式

使用 Application Default Credentials,
通过指定 JSON 格式的密钥文件来使用服务账号。密钥可以指定为密钥文件路径 (Keyfile Path)、密钥内容 (Keyfile JSON) 或 Secret Manager 中的 Secret (Keyfile secret name`). 每次只能使用一种方式定义密钥。如果您需要管理多个密钥，则应配置多个连接。


通过指定有效 凭据配置文件 的路径或内容。凭据配置文件是一个通常包含非敏感元数据的文件，用于指导 google-auth 库如何检索外部主题令牌并将其交换为服务账号访问令牌。

警告
可能需要额外权限

使用 Secret Manager 中密钥的连接要求 Application Default Credentials (ADC) 具有访问 Secret 内容的权限。

注意
另一种存储连接的方式

除了仅在 Secret Manager 中存储密钥外，还可以选择存储整个连接。更多详情请参阅 Google Secret Manager 后端。



默认连接 ID¶
所有与 Google Cloud 相关的 Hook 和 Operator 默认使用 google_cloud_default。


关于 Application Default Credentials 的注意事项¶
在 Google Compute Engine 上运行 Airflow 时，Application Default Credentials 由 GCE 元数据服务器推断；在 GKE 上运行时，由 GKE 元数据服务器推断，这允许将 Kubernetes Service Accounts 映射到 GCP 服务账号 Workload Identity。这在单个 GKE 集群上管理多个具有不同 IAM 足迹的 Airflow 实例的最低权限时非常有用。只需为您的 worker / webserver 部署分配 KSA，Workload Identity 就会将它们映射到单独的 GCP 服务账号（而不是共享集群级 GCE 服务账号）。从安全角度看，它的好处是不将 Google 服务账号密钥存储在磁盘或 Airflow 数据库中，从而避免敏感的长期凭据密钥材料泄露。
从 Airflow 的角度来看，可以通过指定一个空 URI 来使用 Application Default Credentials 进行连接。
例如
export AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT='google-cloud-platform://'




配置连接¶

项目 ID（可选）
要连接的 Google Cloud 项目 ID。它被使用它的 Operator 用作默认项目 ID，并且通常可以在 Operator 级别覆盖。

密钥文件路径
磁盘上 服务账号 密钥文件（JSON 格式）的路径。
如果使用 Application Default Credentials 则无需提供。

密钥文件 JSON 内容
磁盘上 服务账号 密钥文件（JSON 格式）的内容。
如果使用 Application Default Credentials 则无需提供。

凭据配置文件
凭据配置文件的 JSON 内容或文件系统上的凭据配置文件路径。
如果使用 Application Default Credentials 则无需提供。

存储密钥文件 JSON 内容的 Secret 名称
Secret Manager 中包含 服务账号 密钥的 Secret 名称。
如果使用 Application Default Credentials 则无需提供。

范围（逗号分隔）
要用于身份认证的逗号分隔的 Google Cloud 范围 列表。

重试次数
整数，使用随机指数退避算法重试的次数。如果所有重试均失败，googleapiclient.errors.HttpError 表示最后一次请求的错误。如果为零（默认），则只尝试一次请求。

模拟链
可选的服务账号，用于使用短期凭据进行模拟；或者是一系列按链式排列的账号列表，用于获取列表中最后一个账号的 access_token，该账号将在利用此连接的所有请求中被模拟。如果设置为字符串，则该账号必须授予发起账号 Service Account Token Creator IAM 角色。如果设置为逗号分隔的列表，则列表中的身份必须将 Service Account Token Creator IAM 角色授予紧邻的前一个身份，列表中第一个账号将此角色授予发起账号。
在环境变量中指定连接时，应使用 URI 语法，并遵循以下要求


scheme 部分应等于 google-cloud-platform（注意：查找连字符）
authority（用户名、密码、主机、端口）、路径被忽略
query parameters 包含此连接类型特有的信息。接受以下键

project - 项目 ID
key_path - 密钥文件路径
keyfile_dict - 密钥文件 JSON 内容
key_secret_name - 存储密钥文件 JSON 内容的 Secret 名称
key_secret_project_id - 存储密钥文件 JSON 内容的项目 ID
scope - 范围
num_retries - 重试次数




注意，URI 的所有组件都应该进行 URL 编码。
例如，使用 URI 格式
export AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT='google-cloud-platform://?key_path=%2Fkeys%2Fkey.json&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcloud-platform&project=airflow&num_retries=5'


并使用 JSON 格式
export AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT='{"conn_type": "google_cloud_platform", "extra": {"key_path": "/keys/key.json", "scope": "https://www.googleapis.com/auth/cloud-platform", "project": "airflow", "num_retries": 5}}'






直接模拟服务账号¶
Google Operator 支持通过 impersonation_chain 参数（对于也与其他云提供商服务通信的 Operator 则是 google_impersonation_chain）直接模拟服务账号。模拟链也可以直接在 Google Cloud 连接上配置，如上所述，但传递给 Operator 的 impersonation_chain 优先。
例如
import os

from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryCreateEmptyDatasetOperator,
)

IMPERSONATION_CHAIN = "impersonated_account@your_project_id.iam.gserviceaccount.com"

create_dataset = BigQueryCreateEmptyDatasetOperator(
    task_id="create-dataset",
    gcp_conn_id="google_cloud_default",
    dataset_id="test_dataset",
    location="southamerica-east1",
    impersonation_chain=IMPERSONATION_CHAIN,
)


为了使此示例生效，账号 impersonated_account 必须授予 google_cloud_default 连接中指定的服务账号 Service Account Token Creator IAM 角色。这将允许生成 impersonated_account 的访问令牌，从而可以使用其权限代表其行事。impersonated_account 甚至无需生成密钥。
对于连接到多个 Google 服务的 Operator，所有 Hook 都使用相同的 impersonation_chain 值（如果适用）。您还可以模拟来自非发起账号项目中的账号。在这种情况下，被模拟账号的项目 ID 将被用作 Operator 逻辑中的默认项目 ID，除非您在连接配置或 Operator 参数中明确指定了项目 ID。
模拟也可以链式使用：如果连接中指定的服务账号被授予对账号 A 的 Service Account Token Creator 角色，并且账号 A 被授予对账号 B 的此角色，那么我们就可以模拟账号 B。
例如，通过以下 terraform 设置…
terraform {
  required_version = "> 0.11.14"
}
provider "google" {
}
variable "project_id" {
  type = "string"
}
resource "google_service_account" "sa_1" {
  account_id   = "impersonation-chain-1"
  project = "${var.project_id}"
}
resource "google_service_account" "sa_2" {
  account_id   = "impersonation-chain-2"
  project = "${var.project_id}"
}
resource "google_service_account" "sa_3" {
  account_id   = "impersonation-chain-3"
  project = "${var.project_id}"
}
resource "google_service_account" "sa_4" {
  account_id   = "impersonation-chain-4"
  project = "${var.project_id}"
}
resource "google_service_account_iam_member" "sa_4_member" {
  service_account_id = "${google_service_account.sa_4.name}"
  role               = "roles/iam.serviceAccountTokenCreator"
  member             = "serviceAccount:${google_service_account.sa_3.email}"
}
resource "google_service_account_iam_member" "sa_3_member" {
  service_account_id = "${google_service_account.sa_3.name}"
  role               = "roles/iam.serviceAccountTokenCreator"
  member             = "serviceAccount:${google_service_account.sa_2.email}"
}
resource "google_service_account_iam_member" "sa_2_member" {
  service_account_id = "${google_service_account.sa_2.name}"
  role               = "roles/iam.serviceAccountTokenCreator"
  member             = "serviceAccount:${google_service_account.sa_1.email}"
}


…我们应该将 Airflow 连接配置为使用 impersonation-chain-1 账号的密钥，并为 impersonation_chain 参数提供以下值…
PROJECT_ID = os.environ.get("TF_VAR_project_id", "your_project_id")
IMPERSONATION_CHAIN = [
    f"impersonation-chain-2@{PROJECT_ID}.iam.gserviceaccount.com",
    f"impersonation-chain-3@{PROJECT_ID}.iam.gserviceaccount.com",
    f"impersonation-chain-4@{PROJECT_ID}.iam.gserviceaccount.com",
]


…然后请求将使用 impersonation-chain-4 账号的权限执行。


域范围委派¶
除了直接模拟服务账号外，一些 Google Operator、Hook 和 Sensor 还支持域范围委派。委派允许用户或服务账号授予另一个服务账号代表其行事的能力。这意味着委派权限的用户或服务账号可以继续访问和管理自己的资源，而被委派的服务账号也可以访问和管理这些资源。
例如
PROJECT_ID = os.environ.get("TF_VAR_project_id", "your_project_id")

SPREADSHEET = {
    "properties": {"title": "Test1"},
    "sheets": [{"properties": {"title": "Sheet1"}}],
}

from airflow.providers.google.suite.operators.sheets import (
    GoogleSheetsCreateSpreadsheetOperator,
)

create_spreadsheet_operator = GoogleSheetsCreateSpreadsheetOperator(
    task_id="create-spreadsheet",
    gcp_conn_id="google_cloud_default",
    spreadsheet=SPREADSHEET,
    impersonation_chain=f"projects/-/serviceAccounts/SA@{PROJECT_ID}.iam.gserviceaccount.com",
)


请注意，虽然目前大多数 Google Operator 和 Hook 支持域范围委派，但其使用应仅限于 Google Workspace (gsuite) 和营销平台 Operator 和 Hook，或通过 GoogleDiscoveryAPI Hook 访问这些服务。在以下使用场景中已弃用

所有 Google Cloud Operator 和 Hook。
Firebase Hook。
涉及 Google Cloud 的所有跨不同 Provider 的传输 Operator，例如：airflow.providers.amazon.aws.transfers.gcs_to_s3.GCSToS3Operator。