- 大数据
- 批处理
- 流处理
- 数据工程
【免费下载链接】beam
Apache Beam is a unified programming model for Batch and Streaming data processing.
本篇指南围绕 Apache Beam 仓库中的.test-infra/terraform/google-cloud-platform/vertex-ai-featurestore模块展开，介绍如何通过基础设施即代码（Infrastructure-as-Code）在 Google Cloud 上自动化创建 Vertex AI Featurestore，包括 Featurestore、Entity Type 与 Feature 的完整定义。读完本文，你将掌握该模块的三个核心文件（variables.tf、featurestore.tf、synthea.tfvars）的结构与用法，能够自行初始化模块、编写*.tfvars变量文件并完成资源申请，并理解其背后的配置语义与监控参数含义。
模块定位：为 Beam 测试提供 ML 特征存储
Apache Beam 的仓库中维护了一套用于测试的基础设施代码（Infrastructure-as-Code），统一存放于 .test-infra/terraform/google-cloud-platform。这些代码使用 Terraform 为需要 Google Cloud 资源的 Apache Beam 测试预置环境，每个子目录负责一类资源。其中 vertex-ai-featurestore 模块专门用于预置一个Vertex AI Featurestore（Google Cloud 面向机器学习特征管理的托管服务）。
从仓库结构看，该模块只包含 4 个文件，职责非常收敛：
README.md：模块使用说明（初始化、变量文件、apply 步骤）；variables.tf：声明模块所需的全部输入变量及其类型约束；featurestore.tf：定义实际要创建的 Google Cloud 资源（服务启用、Featurestore、Entity Type、Feature）；synthea.tfvars：一份开箱即用的示例变量文件，基于 Synthea 合成的医疗数据构建特征存储。
这个模块是为 Beam 的集成测试场景服务的：测试需要“接近生产”的 Google Cloud 环境来验证 Beam 与 Vertex AI / 健康医疗数据管道的集成。它本身不包含 Beam 管道代码，而是负责把测试所依赖的云侧存储资源准备就绪。
环境前提
使用该模块前，需要先满足 google-cloud-platform 目录的 README 中列出的要求：
- Terraform CLI v1.2.0 及以上；
- Google Cloud SDK，并完成
gcloud init与gcloud auth登录； - 一个已启用计费（billing）的 Google Cloud 项目；
- （可选但强烈建议）IntelliJ 或 VS Code 的 Terraform 插件，便于编写与校验
.tf/.tfvars文件。
所有资源的申请都遵循标准的 Terraform 核心工作流（init → plan/apply），因此在任何子目录中操作方式一致。
三步快速上手
1. 初始化 Terraform 模块
进入模块目录并执行初始化，以拉取 Google 与 random 等 provider 插件：
cd .test-infra/terraform/google-cloud-platform/vertex-ai-featurestore terraform init2. 创建*.tfvars变量文件
Terraform 会从*.tfvars文件读取变量值。在模块同目录下创建一个（名字可自定）：
cd .test-infra/terraform/google-cloud-platform/vertex-ai-featurestore touch vars.tfvars然后参考下文“示例变量文件”与“变量说明”两节填充内容。注意：模块中project与region没有默认值，必须在变量文件中显式给出，否则terraform apply会交互式提示输入。
3. 应用模块
执行 apply 并显式指定变量文件：
cd .test-infra/terraform/google-cloud-platform/vertex-ai-featurestore terraform apply -var-file=vars.tfvars对于变量文件中尚未设置的任何变量，Terraform 会逐个提示补全。
输入变量详解（variables.tf）
模块的全部输入由 variables.tf 声明，共三个：
| 变量 | 类型 | 说明 |
|---|---|---|
project | string | 资源将被预置到的 Google Cloud 项目（必填，无默认值） |
region | string | 资源预置所在的 GCP 区域（必填，无默认值） |
featurestore | object | Featurestore 的整体配置，见下方嵌套结构 |
featurestore对象包含三层嵌套配置：
variable "featurestore" { type = object({ // Featurestore 名称前缀。 name_prefix = string // 配置 Featurestore 的节点数（在线服务容量）。 fixed_node_count = number // Entity Type 配置：map 的 key 即 Entity Type 名称。 entity_types = map(object({ // 该 Entity Type 的 features 配置： // map 的 key 是 Feature 名称，value 是数据类型， // 例如 BOOL、STRING、INT64 等。 description = string features = map(string) })) }) }要点说明：
name_prefix：最终 Featurestore 的实际名称会在该前缀基础上追加随机后缀（见下文featurestore.tf分析），用于避免命名冲突。fixed_node_count：在线服务（online serving）的固定节点数，对应online_serving_config的容量。entity_types：以 map 表达多个实体类型；每个实体类型内，features又是 map，键为 Feature 名、值为其值类型。允许的数据类型（如BOOL、STRING、INT64）以 Vertex AI Featurestore API 的ValueType枚举为准。
资源定义剖析（featurestore.tf）
featurestore.tf 完整定义了从 API 启用到 Feature 落地的整条资源链，共 4 类资源，彼此通过depends_on与引用建立依赖。
1. 启用 AI Platform API
provider "google" { project = var.project } resource "google_project_service" "required" { service = "aiplatform.googleapis.com" disable_on_destroy = false }模块首先确保目标项目已启用aiplatform.googleapis.com（Vertex AI 的底层 API）。disable_on_destroy = false表示销毁该模块资源时不会顺带关闭整个项目的 API，避免影响其他依赖此 API 的资源。
2. 生成随机后缀
resource "random_string" "postfix" { length = 6 upper = false special = false }生成 6 位小写字母数字随机串，用于 Featurestore 名称去重。
3. 创建 Featurestore 本体
resource "google_vertex_ai_featurestore" "default" { depends_on = [google_project_service.required] name = "${var.featurestore.name_prefix}_${random_string.postfix.result}" region = var.region online_serving_config { fixed_node_count = var.featurestore.fixed_node_count } }最终名称形如synthea_ab12cd。online_serving_config.fixed_node_count直接对应变量文件中的fixed_node_count，决定在线特征检索服务的吞吐容量。
4. 创建 Entity Type 及其 Feature
Entity Type（实体类型）是特征的逻辑分组，例如“患者”、“订单”。模块用for_each遍历entity_typesmap 为每个实体类型创建资源：
resource "google_vertex_ai_featurestore_entitytype" "entities" { depends_on = [google_project_service.required] for_each = var.featurestore.entity_types name = each.key featurestore = google_vertex_ai_featurestore.default.id description = each.value.description monitoring_config { categorical_threshold_config { value = 0.3 } numerical_threshold_config { value = 0.3 } snapshot_analysis { disabled = false monitoring_interval_days = 1 staleness_days = 21 } } }这里内置了**特征监控（feature monitoring）**配置，是模块的一个关键细节：
- 分类特征阈值（categorical_threshold_config）：分布偏移检测阈值
0.3； - 数值特征阈值（numerical_threshold_config）：同样为
0.3； - 快照分析（snapshot_analysis）：监控未禁用（
disabled = false），监控间隔 1 天（monitoring_interval_days = 1），数据陈旧窗口 21 天（staleness_days = 21），即每日对比最近快照与 21 天前的快照来捕捉漂移。
Entity Type 之下的每个 Feature 通过for_each逐一创建。由于 Terraform 不支持直接在嵌套 map 上展开两层循环，模块先用locals把“实体类型 × 特征”的二维配置拍平成一张表，再构建以实体类型名.特征名为键的 map，最后交给for_each：
locals { features = flatten([ for entitytype_name, entitytype in var.featurestore.entity_types : [ for feature_name, feature_type in entitytype.features : { entitytype_name = entitytype_name feature_name = feature_name feature_type = feature_type } ] ]) features_map = tomap({ for feature in local.features : "${feature["entitytype_name"]}.${feature["feature_name"]}" => feature }) } resource "google_vertex_ai_featurestore_entitytype_feature" "features" { depends_on = [google_project_service.required] for_each = local.features_map name = each.value["feature_name"] entitytype = google_vertex_ai_featurestore_entitytype.entities[each.value["entitytype_name"]].id value_type = each.value["feature_type"] }这段flatten + tomap的写法是理解该模块如何用两重嵌套 map 驱动资源创建的关键：外层entity_types决定 Entity Type，内层features决定每个 Entity Type 下的 Feature，最终每个 Feature 通过索引表达式google_vertex_ai_featurestore_entitytype.entities[...].id挂靠到其所属的 Entity Type 上，value_type则写入声明好的数据类型。
示例变量文件：基于 Synthea 的医疗特征存储
模块自带的 synthea.tfvars 是一份可直接 apply 的完整示例。它描述的场景是：基于 Synthea 生成的合成患者数据，这些数据存放在 Google Cloud FHIR Store 并通过 BigQuery 流式传输（FHIR-BigQuery streaming），最终被建模进 Vertex AI Featurestore，用于医疗相关的机器学习测试。
文件开头固定了区域与 Featurestore 基础配置：
region = "us-central1" featurestore = { name_prefix = "synthea" fixed_node_count = 1 entity_types = { ... } }由于project在文件中未给出，直接运行terraform apply -var-file=synthea.tfvars时，项目 ID 会以交互提示的方式要求你输入。
该示例定义了 3 个 Entity Type，正好演示了BOOL与STRING两类值类型：
conditions：Snomed 编码的活动性疾病
description：按 Snomed 编码标识患者是否患有某疾病；- 特征均为
BOOL类型，如snomed_10509002、snomed_105531004等，覆盖数百个 Snomed 代码； - 语义：Featurestore 以“实体 ID + 时间戳”为索引，反映某时刻患者已知的活动性疾病；
(患者 id, 时间戳, 疾病标志)元组即患者在那一刻的活动性疾病集合。
medications：RxNorm 编码的活动性用药
description：按 RxNorm 编码标识患者是否有活动性用药；- 特征同样全部为
BOOL，如rxnorm_1000126、rxnorm_108515等； - 语义：与 conditions 相同的时点索引思想，
(患者 id, 时间戳, 用药标志)表示该时刻已知的活动性用药。
observations：Loinc 编码的观测与测量值
description：按 Loinc 编码标识患者的观测/测量值；- 特征是
STRING类型（如loinc_10230_1、loinc_10480_2），值为LOW、MID、HIGH等级； - 语义：
(患者 id, 时间戳, 观测值)表示该时刻患者的最新已知观测结果。
这一设计很好地展示了 Featurestore 的建模方式：同一实体（患者）在不同实体类型下维护不同维度的时点特征——疾病标志、用药标志、观测分级——全部以id + timestamp联合索引，便于按时间回溯特征值，为时间敏感型 ML 测试提供语义正确的训练样本。
从本模块出发：如何定制自己的 Featurestore
在理解三个文件后，你可以按以下步骤定制一套自己的特征存储：
- 拷贝或新建
*.tfvars：参照synthea.tfvars的骨架，先设置project、region、name_prefix、fixed_node_count； - 按业务定义
entity_types：每个实体类型给出description，featuresmap 中按特征名 = 数据类型填写，注意值类型必须是 Vertex AI 的ValueType允许取值（BOOL、STRING、INT64、DOUBLE等），例如把snomed_xxx = "BOOL"换成daily_orders = "INT64"； - 在模块目录执行：
cd .test-infra/terraform/google-cloud-platform/vertex-ai-featurestore terraform init terraform apply -var-file=你的文件.tfvars- 验证与清理：apply 完成后可通过
terraform state list查看已创建的资源；测试结束后用terraform destroy -var-file=你的文件.tfvars释放资源（不会影响已启用的项目级 API）。
总结
vertex-ai-featurestore模块是 Apache Beam 测试基础设施中一个典型而完整的 Terraform 示例：它以 3 个输入变量驱动 4 类云资源的创建，用for_each+flatten/tomap技巧将两层嵌套 map 展开为成百上千个 Feature 资源，并内建了特征漂移监控配置。无论是为 Beam 与 Vertex AI 的集成测试搭建环境，还是学习如何在 Terraform 中批量建模 Google Cloud 的层次化资源，这个模块都提供了可直接复用、可本地验证的参考实现。
进一步阅读：模块总览见 .test-infra/terraform/google-cloud-platform/README.md；其余测试基础设施模块（如 google-kubernetes-engine）遵循相同的工作流。
- 大数据
- 批处理
- 流处理
- 数据工程
【免费下载链接】beam
Apache Beam is a unified programming model for Batch and Streaming data processing.
相关推荐
Apache Beam 测试基础设施的 GCP Terraform IaC 实战指南：从私有 GKE 集群到 Vertex AI Featurestore
Apache Beam 测试基础设施的 GCP Terraform IaC 实战指南：从私有 GKE 集群到 Vertex AI Featurestore Ap
大数据批处理流处理数据工程使用 Terraform 在 Google Cloud 上为 Apache Beam 测试搭建私有 GKE 集群
使用 Terraform 在 Google Cloud 上为 Apache Beam 测试搭建私有 GKE 集群 Apache Beam 的端到端测试（如 Ka
大数据批处理流处理数据工程Apache Beam 测试基础设施：基于 Terraform 与 gcloud IAP 隧道搭建 Kafka 私有网络访问代理
Apache Beam 测试基础设施：基于 Terraform 与 gcloud IAP 隧道搭建 Kafka 私有网络访问代理 本文围绕 Apache Bea
大数据批处理流处理数据工程
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
