此模板会创建一个流处理流水线,用于流式传输 Bigtable 数据更改记录,并使用 Dataflow Runner V2 将其写入 Vertex AI Vector Search。
流水线要求
- Bigtable 源实例必须存在。
- Bigtable 源表必须已存在,且 必须启用变更数据流。
- Bigtable 应用配置文件必须存在。
- 向量搜索索引路径必须存在。
模板参数
参数 | 说明 |
---|---|
embeddingColumn |
在其中存储嵌入的完全限定列名。格式为 cf:col。 |
embeddingByteSize |
嵌入数组中每个条目的字节大小。对于 float,请使用 4;对于 double,请使用 8。默认值为 4 。 |
vectorSearchIndex |
将流式传输更改的 Vector Search 索引,格式为“projects/{projectID}/locations/{region}/indexes/{indexID}”(不含前后空格)。例如:projects/123/locations/us-east1/indexes/456 。 |
bigtableChangeStreamAppProfile |
用于区分 Bigtable 中工作负载的应用配置文件。 |
bigtableReadInstanceId |
表所属的 Bigtable 实例的 ID。 |
bigtableReadTableId |
要读取的 Bigtable 表。 |
bigtableMetadataTableTableId |
可选:要创建的元数据表的 ID。如果未设置,Bigtable 会生成一个 ID。 |
crowdingTagColumn |
可选:存储拥挤标记的完全限定列名称,格式为 cf:col 。 |
allowRestrictsMappings |
可选:要用作 allow 限制的列的以逗号分隔完全限定列名及其别名。每个列名称都必须采用 cf:col->alias 格式。 |
denyRestrictsMappings |
可选:要用作 deny 限制的列的以逗号分隔的完全限定列名称及其别名。每个列名称都必须采用 cf:col->alias 格式。 |
intNumericRestrictsMappings |
可选:要用作整数 numeric_restricts 的列的以逗号分隔完全限定列名及其别名。每个列名称都必须采用 cf:col->alias 格式。 |
floatNumericRestrictsMappings |
可选:要用作浮点数(4 字节)numeric_restricts 的列的以逗号分隔完全限定列名及其别名。每个列名称都必须采用 cf:col->alias 格式 |
doubleNumericRestrictsMappings |
可选:要用作双精度(8 字节)numeric_restricts 的列的英文逗号分隔的完全限定列名称及其别名。每个列名称都必须采用 cf:col->alias 格式 |
upsertMaxBatchSize |
可选:在将批次更新/插入向量搜索索引之前要缓冲的更新/插入数量上限。当有 upsertBatchSize 条记录准备就绪时,系统会发送批量数据。
示例:10 。 |
upsertMaxBufferDuration |
可选:将一批更新/插入操作发送到向量搜索之前的最大延迟。当有 upsertBatchSize 条记录准备就绪时,系统会发送批量数据。允许的格式如下:Ns (表示秒,例如:5s)、Nm (表示分钟,例如 12m)、Nh (表示小时,例如 2h)。默认值:10s 。 |
deleteMaxBatchSize |
可选:从向量搜索索引中删除批次之前要缓冲的最大删除次数。当有 deleteBatchSize 条记录准备就绪时,系统会发送批量数据。例如:10 。 |
deleteMaxBufferDuration |
可选:在向 Vector Search 发送批量删除操作之前的最大延迟时间。当有 deleteBatchSize 条记录准备就绪时,系统会发送批量数据。允许的格式如下:Ns (表示秒,例如:5s)、Nm (表示分钟,例如 12m)、Nh (表示小时,例如 2h)。默认值:10s 。 |
dlqDirectory |
可选:用于存储未处理记录以及无法处理原因的路径。默认值为 Dataflow 作业的临时位置下的目录。默认值适用于大多数情况。 |
bigtableChangeStreamMetadataInstanceId |
可选:要用于变更数据流连接器元数据表的 Bigtable 实例。默认值为空。 |
bigtableChangeStreamMetadataTableTableId |
可选:要使用的 Bigtable 变更数据流连接器元数据表 ID。如果未提供,系统会在流水线流期间自动创建 Bigtable 变更数据流连接器元数据表。默认值为空。 |
bigtableChangeStreamCharset |
可选:读取值和列限定符时的 Bigtable 变更数据流字符集名称。默认值为 UTF-8。 |
bigtableChangeStreamStartTimestamp |
可选:要用于读取变更数据流的起始 DateTime(含边界值)(https://tools.ietf.org/html/rfc3339)。例如,2022-05-05T07:59:59Z。默认为流水线启动时的时间戳。 |
bigtableChangeStreamIgnoreColumnFamilies |
可选:系统不会捕获对列族名称所做更改的逗号分隔列表。默认值为空。 |
bigtableChangeStreamIgnoreColumns |
可选:要忽略的列名称变更的英文逗号分隔列表。默认值为空。 |
bigtableChangeStreamName |
可选:客户端流水线的唯一名称。此参数允许您从先前运行的流水线停止的位置继续处理。默认为自动生成的名称。如需了解所用的值,请参阅 Dataflow 作业日志。 |
bigtableChangeStreamResume |
可选:设置为 true 时,新流水线将从具有相同名称的先前运行的流水线停止时的点开始处理。如果具有该名称的流水线过去从未运行过,则新流水线将无法启动。
使用 如果设置为 false,系统会启动新的流水线。如果以给定来源为来源,过去已运行过与 默认值为 false。 |
bigtableReadProjectId |
可选:要从中读取 Bigtable 数据的项目。此参数的默认项目是 Dataflow 流水线在其中运行的项目。 |
运行模板
控制台
- 转到 Dataflow 基于模板创建作业页面。 转到“基于模板创建作业”
- 在作业名称字段中,输入唯一的作业名称。
- 可选:对于区域性端点,从下拉菜单中选择一个值。默认区域为
us-central1
。如需查看可以在其中运行 Dataflow 作业的区域列表,请参阅 Dataflow 位置。
- 从 Dataflow 模板下拉菜单中,选择 the Bigtable Change Streams to Vector Search template。
- 在提供的参数字段中,输入您的参数值。
- 点击运行作业。
gcloud CLI
在 shell 或终端中,运行模板:
gcloud dataflow flex-template run JOB_NAME \ --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_Vector_Search \ --project=PROJECT_ID \ --region=REGION_NAME \ --parameters \ embeddingColumn=EMBEDDING_COLUMN,\ embeddingByteSize=EMBEDDING_BYTE_SIZE,\ vectorSearchIndex=VECTOR_SEARCH_INDEX,\ bigtableChangeStreamAppProfile=BIGTABLE_CHANGE_STREAM_APP_PROFILE,\ bigtableReadInstanceId=BIGTABLE_READ_INSTANCE_ID,\ bigtableReadTableId=BIGTABLE_READ_TABLE_ID,\
替换以下内容:
JOB_NAME
:您选择的唯一性作业名称VERSION
:您要使用的模板的版本您可使用以下值:
latest
,以使用模板的最新版本,该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates-REGION_NAME/latest/) 中可用- 版本名称(如
2023-09-12-00_RC00
),以使用模板的特定版本,该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates-REGION_NAME/) 中
REGION_NAME
:要在其中部署 Dataflow 作业的区域,例如us-central1
EMBEDDING_COLUMN
:嵌入列EMBEDDING_BYTE_SIZE
:嵌入数组的字节大小。可以是 4 或 8。VECTOR_SEARCH_INDEX
:向量搜索索引路径BIGTABLE_CHANGE_STREAM_APP_PROFILE
:Bigtable 应用配置文件 IDBIGTABLE_READ_INSTANCE_ID
:源 Bigtable 实例 IDBIGTABLE_READ_TABLE_ID
:源 Bigtable 表 ID
API
如需使用 REST API 来运行模板,请发送 HTTP POST 请求。如需详细了解 API 及其授权范围,请参阅 projects.templates.launch
。
POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch { "launchParameter": { "jobName": "JOB_NAME", "parameters": { "embeddingColumn": "EMBEDDING_COLUMN", "embeddingByteSize": "EMBEDDING_BYTE_SIZE", "vectorSearchIndex": "VECTOR_SEARCH_INDEX", "bigtableChangeStreamAppProfile": "BIGTABLE_CHANGE_STREAM_APP_PROFILE", "bigtableReadInstanceId": "BIGTABLE_READ_INSTANCE_ID", "bigtableReadTableId": "BIGTABLE_READ_TABLE_ID", }, "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Bigtable_Change_Streams_to_Vector_Search", "environment": { "maxWorkers": "10" } } }
替换以下内容:
PROJECT_ID
:您要在其中运行 Dataflow 作业的 Google Cloud 项目的 IDJOB_NAME
:您选择的唯一性作业名称VERSION
:您要使用的模板的版本您可使用以下值:
latest
,以使用模板的最新版本,该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates-REGION_NAME/latest/) 中可用- 版本名称(如
2023-09-12-00_RC00
),以使用模板的特定版本,该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates-REGION_NAME/) 中
LOCATION
:要在其中部署 Dataflow 作业的区域,例如us-central1
EMBEDDING_COLUMN
:嵌入列EMBEDDING_BYTE_SIZE
:嵌入数组的字节大小。可以是 4 或 8。VECTOR_SEARCH_INDEX
:向量搜索索引路径BIGTABLE_CHANGE_STREAM_APP_PROFILE
:Bigtable 应用配置文件 IDBIGTABLE_READ_INSTANCE_ID
:源 Bigtable 实例 IDBIGTABLE_READ_TABLE_ID
:源 Bigtable 表 ID