Streaming perubahan Bigtable ke template Vector Search

Template ini membuat pipeline streaming untuk melakukan streaming data perubahan Bigtable dan menulisnya 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.
• Jalur indeks Vector Search harus ada.

Parameter template

Parameter	Deskripsi
embeddingColumn	Nama kolom yang sepenuhnya memenuhi syarat tempat penyematan disimpan. Dalam format cf:col.
embeddingByteSize	Ukuran byte setiap entri dalam array penyematan. Gunakan 4 untuk float dan 8 untuk double. Nilai defaultnya adalah 4.
vectorSearchIndex	Indeks Penelusuran Vektor tempat perubahan akan di-streaming, dalam format 'projects/{projectID}/locations/{region}/indexes/{indexID}' (tanpa spasi di awal atau di akhir). Contoh: projects/123/locations/us-east1/indexes/456.
bigtableChangeStreamAppProfile	Profil aplikasi yang digunakan untuk membedakan beban kerja di Bigtable.
bigtableReadInstanceId	ID instance Bigtable yang berisi tabel.
bigtableReadTableId	Tabel Bigtable yang akan dibaca.
bigtableMetadataTableTableId	Opsional: ID untuk tabel metadata yang dibuat. Jika tidak ditetapkan, Bigtable akan membuat ID.
crowdingTagColumn	Opsional: Nama kolom yang sepenuhnya memenuhi syarat tempat tag kepadatan disimpan, dalam format cf:col.
allowRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai allow membatasi, ditambah aliasnya. Setiap nama kolom harus dalam format cf:col->alias.
denyRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai deny membatasi, ditambah aliasnya. Setiap nama kolom harus dalam format cf:col->alias.
intNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai bilangan bulat numeric_restricts, ditambah aliasnya. Setiap nama kolom harus dalam format cf:col->alias.
floatNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai float (4 byte) numeric_restricts, ditambah aliasnya. Setiap nama kolom harus dalam format cf:col->alias
doubleNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai numeric_restricts ganda (8 byte), ditambah aliasnya. Setiap nama kolom harus dalam format cf:col->alias
upsertMaxBatchSize	Opsional: Jumlah maksimum pembaruan dan penyisipan untuk di-buffer sebelum memperbarui dan menyisipkan batch ke indeks Penelusuran Vektor. Batch dikirim saat ada upsertBatchSize data yang siap. Contoh: 10.
upsertMaxBufferDuration	Opsional: Penundaan maksimum sebelum batch pembaruan dan penyisipan dikirim ke Vector Search. Batch dikirim saat ada upsertBatchSize data yang siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5d), Nm untuk menit (contoh: 12m), dan Nh untuk jam (contoh: 2h). Default: 10s.
deleteMaxBatchSize	Opsional: Jumlah maksimum penghapusan yang akan di-buffer sebelum menghapus batch dari indeks Penelusuran Vektor. Batch dikirim saat ada deleteBatchSize data yang siap. Contoh: 10.
deleteMaxBufferDuration	Opsional: Penundaan maksimum sebelum batch penghapusan dikirim ke Vector Search. Batch dikirim saat ada data deleteBatchSize yang siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5d), Nm untuk menit (contoh: 12m), dan Nh untuk jam (contoh: 2h). Default: 10s.
dlqDirectory	Opsional: Jalur untuk menyimpan data yang belum diproses dengan alasan kegagalan pemrosesannya. Default adalah direktori di bawah lokasi sementara tugas Dataflow. Nilai default sesuai untuk sebagian besar skenario.
bigtableChangeStreamMetadataInstanceId	Opsional: Instance Bigtable yang akan digunakan untuk tabel metadata konektor aliran data perubahan. Default-nya adalah kosong.
bigtableChangeStreamMetadataTableTableId	Opsional: ID tabel metadata konektor aliran perubahan Bigtable yang akan digunakan. Jika tidak diberikan, tabel metadata konektor aliran perubahan Bigtable akan otomatis dibuat selama alur pipeline. Default-nya adalah kosong.
bigtableChangeStreamCharset	Opsional: Bigtable mengubah nama set karakter aliran saat membaca nilai dan penentu kolom. Setelan defaultnya adalah UTF-8.
bigtableChangeStreamStartTimestamp	Opsional: DateTime awal, inklusif, yang akan digunakan untuk membaca aliran perubahan (https://tools.ietf.org/html/rfc3339). Misalnya, 2022-05-05T07:59:59Z. Default-nya adalah stempel waktu saat pipeline dimulai.
bigtableChangeStreamIgnoreColumnFamilies	Opsional: Daftar nama keluarga kolom yang dipisahkan koma yang akan diubah dan tidak akan dicatat. Default-nya adalah kosong.
bigtableChangeStreamIgnoreColumns	Opsional: Daftar nama kolom yang dipisahkan koma yang akan diubah dan tidak akan dicatat. Default-nya adalah kosong.
bigtableChangeStreamName	Opsional: Nama unik untuk pipeline klien. Parameter ini memungkinkan Anda melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan dihentikan. Secara default, nama yang dibuat secara otomatis. Lihat log tugas Dataflow untuk mengetahui nilai yang digunakan.
bigtableChangeStreamResume	Opsional: Jika ditetapkan ke benar, pipeline baru akan melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan dengan nama yang sama dihentikan. Jika pipeline dengan nama tersebut tidak pernah berjalan sebelumnya, pipeline baru akan gagal dimulai. Gunakan parameter bigtableChangeStreamName untuk menentukan baris pipeline. Jika disetel ke salah, pipeline baru akan dimulai. Jika pipeline dengan nama yang sama seperti bigtableChangeStreamName telah berjalan sebelumnya untuk sumber tertentu, pipeline baru akan gagal dimulai. Nilai defaultnya adalah salah (false).
bigtableReadProjectId	Opsional: Project tempat membaca data Bigtable. Default untuk parameter ini adalah project tempat pipeline Dataflow berjalan.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat 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 defaultnya adalah us-central1.

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

4. Dari menu drop-down Dataflow template, 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, 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 template versi 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 dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error 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 penyematan. 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 POST HTTP. 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 project Google Cloud 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 template versi 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 dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error 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 penyematan. 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);
  }
}