Streaming perubahan Bigtable ke template Vector Search

Template ini membuat pipeline streaming untuk melakukan streaming data Bigtable yang mengubah rekaman dan menuliskannya ke Vertex AI Vector Search menggunakan Dataflow Runner V2.

Persyaratan pipeline

• Instance sumber Bigtable harus ada.
• Tabel sumber Bigtable harus ada, dan tabel harus mengaktifkan aliran perubahan.
• Profil aplikasi Bigtable harus ada.
• Path indeks Vector Search harus ada.

Parameter template

Parameter yang diperlukan

• embeddingColumn: Nama kolom yang sepenuhnya memenuhi syarat tempat embedding disimpan. Dalam format cf:col.
• embeddingByteSize: Ukuran byte setiap entri dalam array embedding. Gunakan 4 untuk Float, dan 8 untuk Double. Default: 4.
• vectorSearchIndex: Indeks Vector Search tempat perubahan akan di-streaming, dalam format 'projects/{projectID}/locations/{region}/indexes/{indexID}' (tanpa spasi di awal atau akhir) Misalnya, projects/123/locations/us-east1/indexes/456.
• bigtableChangeStreamAppProfile: ID profil aplikasi Bigtable. Profil aplikasi harus menggunakan perutean cluster tunggal dan mengizinkan transaksi baris tunggal.
• bigtableReadInstanceId: ID instance Bigtable sumber.
• bigtableReadTableId: ID tabel Bigtable sumber.

Parameter opsional

• bigtableMetadataTableTableId: ID Tabel yang digunakan untuk membuat tabel metadata.
• crowdingTagColumn: Nama kolom yang sepenuhnya memenuhi syarat tempat tag keramaian disimpan. Dalam format cf:col.
• allowRestrictsMappings: Nama kolom yang memenuhi syarat sepenuhnya yang dipisahkan koma dari kolom yang harus digunakan sebagai pembatasan allow, dengan aliasnya. Dalam format cf:col->alias.
• denyRestrictsMappings: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang harus digunakan sebagai pembatasan deny, dengan aliasnya. Dalam format cf:col->alias.
• intNumericRestrictsMappings: Nama kolom yang sepenuhnya memenuhi syarat yang dipisahkan koma dari kolom yang harus digunakan sebagai numeric_restricts bilangan bulat, dengan aliasnya. Dalam format cf:col->alias.
• floatNumericRestrictsMappings: Nama kolom yang sepenuhnya memenuhi syarat yang dipisahkan koma dari kolom yang harus digunakan sebagai numeric_restricts float (4 byte), dengan aliasnya. Dalam format cf:col->alias.
• doubleNumericRestrictsMappings: Nama kolom yang sepenuhnya memenuhi syarat yang dipisahkan koma dari kolom yang harus digunakan sebagai numeric_restricts ganda (8 byte), dengan aliasnya. Dalam format cf:col->alias.
• upsertMaxBatchSize: Jumlah maksimum upsert yang akan di-buffer sebelum meng-upsert batch ke Indeks Vector Search. Batch akan dikirim saat ada rekaman upsertBatchSize yang siap, atau waktu tunggu upsertBatchDelay untuk rekaman apa pun telah berlalu. Contoh, 10. Defaultnya adalah: 10.
• upsertMaxBufferDuration: Penundaan maksimum sebelum batch upsert dikirim ke Vector Search.Batch akan dikirim saat ada upsertBatchSize rekaman yang siap, atau waktu tunggu upsertBatchDelay telah berlalu untuk rekaman apa pun. Format yang diizinkan adalah: Ns (untuk detik, contoh: 5s), Nm (untuk menit, contoh: 12m), Nh (untuk jam, contoh: 2h). Contoh, 10s. Nilai defaultnya adalah: 10 detik.
• deleteMaxBatchSize: Jumlah maksimum penghapusan yang akan di-buffer sebelum menghapus batch dari Indeks Penelusuran Vektor. Batch akan dikirim saat ada deleteBatchSize data yang siap, atau waktu tunggu deleteBatchDelay untuk data apa pun telah berlalu. Contoh, 10. Defaultnya adalah: 10.
• deleteMaxBufferDuration: Penundaan maksimum sebelum batch penghapusan dikirim ke Vector Search.Batch akan dikirim saat ada deleteBatchSize rekaman yang siap, atau waktu deleteBatchDelay telah berlalu untuk rekaman apa pun yang menunggu. Format yang diizinkan adalah: Ns (untuk detik, contoh: 5s), Nm (untuk menit, contoh: 12m), Nh (untuk jam, contoh: 2h). Contoh, 10s. Nilai defaultnya adalah: 10 detik.
• dlqDirectory: Jalur untuk menyimpan semua catatan yang belum diproses beserta alasan kegagalan pemrosesannya. Defaultnya adalah direktori di bawah lokasi sementara tugas Dataflow. Nilai default sudah cukup dalam sebagian besar kondisi.
• bigtableChangeStreamMetadataInstanceId: ID instance metadata aliran perubahan Bigtable. Nilai defaultnya adalah kosong.
• bigtableChangeStreamMetadataTableTableId: ID tabel metadata konektor aliran perubahan data Bigtable. Jika tidak disediakan, tabel metadata konektor aliran perubahan Bigtable akan otomatis dibuat selama eksekusi pipeline. Nilai defaultnya adalah kosong.
• bigtableChangeStreamCharset: Nama charset aliran perubahan Bigtable. Nilai defaultnya adalah: UTF-8.
• bigtableChangeStreamStartTimestamp: Stempel waktu awal (https://tools.ietf.org/html/rfc3339), inklusif, yang akan digunakan untuk membaca aliran perubahan. Misalnya, 2022-05-05T07:59:59Z. Default-nya adalah stempel waktu waktu mulai pipeline.
• bigtableChangeStreamIgnoreColumnFamilies: Daftar perubahan nama grup kolom yang dipisahkan koma untuk diabaikan. Nilai defaultnya adalah kosong.
• bigtableChangeStreamIgnoreColumns: Daftar perubahan nama kolom yang dipisahkan koma untuk diabaikan. Contoh: "cf1:col1,cf2:col2". Nilai defaultnya adalah kosong.
• bigtableChangeStreamName: Nama unik untuk pipeline klien. Memungkinkan Anda melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan berhenti. Secara default, nama dibuat secara otomatis. Lihat log tugas Dataflow untuk nilai yang digunakan.
• bigtableChangeStreamResume: Jika disetel ke true, pipeline baru akan melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan dengan nilai bigtableChangeStreamName yang sama berhenti. Jika pipeline dengan nilai bigtableChangeStreamName yang diberikan belum pernah dijalankan, pipeline baru tidak akan dimulai. Jika disetel ke false, pipeline baru akan dimulai. Jika pipeline dengan nilai bigtableChangeStreamName yang sama telah berjalan untuk sumber tertentu, pipeline baru tidak akan dimulai. Nilai defaultnya adalah false.
• bigtableReadChangeStreamTimeoutMs: Waktu tunggu untuk permintaan Bigtable ReadChangeStream dalam milidetik.
• bigtableReadProjectId: Project ID Bigtable. Defaultnya adalah project untuk tugas Dataflow.

Menjalankan template

Konsol

1. Buka halaman Dataflow Create job from template.
Buka Membuat tugas dari template
2. Di kolom Nama tugas, masukkan nama tugas yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Template Dataflow, pilih the Bigtable Change Streams to Vector Search template.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Run job.

gcloud CLI

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

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,\

Ganti kode berikut:

• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi template terbaru, yang tersedia di folder induk tanpa tanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk yang diberi tanggal di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang merusak ini memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• EMBEDDING_COLUMN: kolom Embedding
• EMBEDDING_BYTE_SIZE: Ukuran byte array embedding. Dapat berupa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Vector Search
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: ID profil aplikasi Bigtable
• BIGTABLE_READ_INSTANCE_ID: ID Instance Bigtable sumber
• BIGTABLE_READ_TABLE_ID: ID tabel Bigtable sumber

API

Untuk menjalankan template menggunakan REST API, kirim permintaan HTTP POST. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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" }
  }
}

Ganti kode berikut:

• PROJECT_ID: ID Google Cloud project tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi template terbaru, yang tersedia di folder induk tanpa tanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk yang diberi tanggal di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang merusak ini memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• EMBEDDING_COLUMN: kolom Embedding
• EMBEDDING_BYTE_SIZE: Ukuran byte array embedding. Dapat berupa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Vector Search
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: ID profil aplikasi Bigtable
• BIGTABLE_READ_INSTANCE_ID: ID Instance Bigtable sumber
• BIGTABLE_READ_TABLE_ID: ID tabel Bigtable sumber

Kode sumber template

Java

/*
 * Copyright (C) 2024 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates.bigtablechangestreamstovectorsearch;

import com.google.cloud.Timestamp;
import com.google.cloud.aiplatform.v1.IndexDatapoint;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadChangeStreamOptions;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadOptions;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamsToVectorSearchOptions;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.DurationUtils;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.GroupIntoBatches;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.transforms.WithKeys;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Template(
    name = "Bigtable_Change_Streams_to_Vector_Search",
    category = TemplateCategory.STREAMING,
    displayName = "Bigtable Change Streams to Vector Search",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into Vertex AI Vector Search using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamsToVectorSearchOptions.class,
    optionsOrder = {
      BigtableChangeStreamsToVectorSearchOptions.class,
      ReadChangeStreamOptions.class,
      ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigtable-change-streams-to-vector-search",
    flexContainerName = "bigtable-changestreams-to-vector-search",
    contactInformation = "https://cloud.google.com/support",
    streaming = true)
public final class BigtableChangeStreamsToVectorSearch {
  private static final Logger LOG =
      LoggerFactory.getLogger(BigtableChangeStreamsToVectorSearch.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  /**
   * Main entry point for executing the pipeline.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) throws Exception {
    LOG.info("Starting replication from Cloud Bigtable Change Streams to Vector Search");

    BigtableChangeStreamsToVectorSearchOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigtableChangeStreamsToVectorSearchOptions.class);

    run(options);
  }

  public static PipelineResult run(BigtableChangeStreamsToVectorSearchOptions options)
      throws IOException {
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    boolean hasUseRunnerV2 = false;
    for (String experiment : experiments) {
      if (experiment.equalsIgnoreCase(USE_RUNNER_V2_EXPERIMENT)) {
        hasUseRunnerV2 = true;
        break;
      }
    }
    if (!hasUseRunnerV2) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);

    Instant startTimestamp =
        options.getBigtableChangeStreamStartTimestamp().isEmpty()
            ? Instant.now()
            : toInstant(Timestamp.parseTimestamp(options.getBigtableChangeStreamStartTimestamp()));

    String bigtableProjectId = getBigtableProjectId(options);

    LOG.info("  - startTimestamp {}", startTimestamp);
    LOG.info("  - bigtableReadInstanceId {}", options.getBigtableReadInstanceId());
    LOG.info("  - bigtableReadTableId {}", options.getBigtableReadTableId());
    LOG.info("  - bigtableChangeStreamAppProfile {}", options.getBigtableChangeStreamAppProfile());
    LOG.info("  - embeddingColumn {}", options.getEmbeddingColumn());
    LOG.info("  - crowdingTagColumn {}", options.getCrowdingTagColumn());
    LOG.info("  - project {}", options.getProject());
    LOG.info("  - indexName {}", options.getVectorSearchIndex());

    String indexName = options.getVectorSearchIndex();

    String vertexRegion = Utils.extractRegionFromIndexName(indexName);
    String vertexEndpoint = vertexRegion + "-aiplatform.googleapis.com:443";

    final Pipeline pipeline = Pipeline.create(options);

    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? BigtableIO.ExistingPipelineOptions.RESUME_OR_FAIL
                    : BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS)
            .withProjectId(bigtableProjectId)
            .withAppProfileId(options.getBigtableChangeStreamAppProfile())
            .withInstanceId(options.getBigtableReadInstanceId())
            .withTableId(options.getBigtableReadTableId())
            .withMetadataTableInstanceId(options.getBigtableChangeStreamMetadataInstanceId())
            .withMetadataTableTableId(options.getBigtableMetadataTableTableId())
            .withStartTime(startTimestamp);

    PCollectionTuple results =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply("Create Values", Values.create())
            .apply(
                "Converting to Vector Search Datapoints",
                ParDo.of(
                        new ChangeStreamMutationToDatapointOperationFn(
                            options.getEmbeddingColumn(),
                            options.getEmbeddingByteSize(),
                            options.getCrowdingTagColumn(),
                            Utils.parseColumnMapping(options.getAllowRestrictsMappings()),
                            Utils.parseColumnMapping(options.getDenyRestrictsMappings()),
                            Utils.parseColumnMapping(options.getIntNumericRestrictsMappings()),
                            Utils.parseColumnMapping(options.getFloatNumericRestrictsMappings()),
                            Utils.parseColumnMapping(options.getDoubleNumericRestrictsMappings())))
                    .withOutputTags(
                        ChangeStreamMutationToDatapointOperationFn.UPSERT_DATAPOINT_TAG,
                        TupleTagList.of(
                            ChangeStreamMutationToDatapointOperationFn.REMOVE_DATAPOINT_TAG)));
    results
        .get(ChangeStreamMutationToDatapointOperationFn.UPSERT_DATAPOINT_TAG)
        .apply("Add placeholer keys", WithKeys.of("placeholder"))
        .apply(
            "Batch Contents",
            GroupIntoBatches.<String, IndexDatapoint>ofSize(
                    bufferSizeOption(options.getUpsertMaxBatchSize()))
                .withMaxBufferingDuration(
                    bufferDurationOption(options.getUpsertMaxBufferDuration())))
        .apply("Map to Values", Values.create())
        .apply(
            "Upsert Datapoints to VectorSearch",
            ParDo.of(new UpsertDatapointsFn(vertexEndpoint, indexName)))
        .apply(
            "Write errors to DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    results
        .get(ChangeStreamMutationToDatapointOperationFn.REMOVE_DATAPOINT_TAG)
        .apply("Add placeholder keys", WithKeys.of("placeholer"))
        .apply(
            "Batch Contents",
            GroupIntoBatches.<String, String>ofSize(
                    bufferSizeOption(options.getDeleteMaxBatchSize()))
                .withMaxBufferingDuration(
                    bufferDurationOption(options.getDeleteMaxBufferDuration())))
        .apply("Map to Values", Values.create())
        .apply(
            "Remove Datapoints From VectorSearch",
            ParDo.of(new RemoveDatapointsFn(vertexEndpoint, indexName)))
        .apply(
            "Write errors to DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static String getBigtableProjectId(BigtableChangeStreamsToVectorSearchOptions options) {
    return StringUtils.isEmpty(options.getBigtableReadProjectId())
        ? options.getProject()
        : options.getBigtableReadProjectId();
  }

  private static Instant toInstant(Timestamp timestamp) {
    if (timestamp == null) {
      return null;
    } else {
      return Instant.ofEpochMilli(timestamp.getSeconds() * 1000 + timestamp.getNanos() / 1000000);
    }
  }

  private static int bufferSizeOption(int size) {
    if (size < 1) {
      size = 1;
    }

    return size;
  }

  private static Duration bufferDurationOption(String duration) {
    if (duration.isEmpty()) {
      return Duration.standardSeconds(1);
    }

    return DurationUtils.parseDuration(duration);
  }

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamsToVectorSearchOptions options) {
    String dlqDirectory = options.getDlqDirectory();
    if (dlqDirectory.isEmpty()) {
      LOG.info("Falling back to temp dir for DLQ");

      String tempLocation = options.as(DataflowPipelineOptions.class).getTempLocation();

      LOG.info("Have temp location {}", tempLocation);
      if (tempLocation == null || tempLocation.isEmpty()) {
        tempLocation = "/";
      } else if (!tempLocation.endsWith("/")) {
        tempLocation += "/";
      }

      dlqDirectory = tempLocation + "dlq";
    }

    LOG.info("Writing dead letter queue to: {}", dlqDirectory);

    return DeadLetterQueueManager.create(dlqDirectory, 1);
  }
}