Bigtable mengubah aliran data ke template Vector Search

Template ini membuat pipeline streaming untuk melakukan streaming kumpulan data perubahan data 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 tersebut harus mengaktifkan aliran data perubahan.
• Profil aplikasi Bigtable harus ada.
• Jalur indeks Penelusuran Vektor harus ada.

Parameter template

Parameter	Deskripsi
embeddingColumn	Nama kolom yang sepenuhnya memenuhi syarat tempat embeddings disimpan. Dalam format cf:col.
embeddingByteSize	Ukuran byte setiap entri dalam array embedding. Gunakan 4 untuk {i>float<i} dan 8 untuk {i>double<i}. Nilai defaultnya adalah 4.
vectorSearchIndex	Indeks Vector Search tempat perubahan akan di-streaming, dalam format 'projects/{projectID}/locations/{region}/indexes/{indexID}' (tanpa spasi di bagian awal atau 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 untuk dibaca.
bigtableMetadataTableTableId	Opsional: ID untuk tabel metadata yang dibuat. Jika tidak disetel, Bigtable akan menghasilkan ID.
crowdingTagColumn	Opsional: Nama kolom yang sepenuhnya memenuhi syarat tempat tag kerumunan disimpan, dalam format cf:col.
allowRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai pembatasan allow, serta aliasnya. Setiap nama kolom harus menggunakan format cf:col->alias.
denyRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai pembatasan deny, serta aliasnya. Setiap nama kolom harus menggunakan format cf:col->alias.
intNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai bilangan bulat numeric_restricts, beserta aliasnya. Setiap nama kolom harus menggunakan format cf:col->alias.
floatNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai numeric_restricts float (4 byte), beserta aliasnya. Setiap nama kolom harus dalam format cf:col->alias
doubleNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai numeric_restricts ganda (8 byte) beserta aliasnya. Setiap nama kolom harus dalam format cf:col->alias
upsertMaxBatchSize	Opsional: Jumlah maksimum pembaruan dan penyisipan yang akan di-buffer sebelum pembaruan dan penyisipan batch ke indeks Vector Search. Batch dikirim jika ada data upsertBatchSize yang siap. Contoh: 10.
upsertMaxBufferDuration	Opsional: Penundaan maksimum sebelum sekumpulan pembaruan dan penyisipan dikirim ke Vector Search. Batch dikirim jika sudah ada data upsertBatchSize yang siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5 detik), Nm untuk menit (contoh: 12 menit), dan Nh untuk jam (contoh: 2 jam). Default: 10s.
deleteMaxBatchSize	Opsional: Jumlah maksimum penghapusan yang akan di-buffer sebelum menghapus batch dari indeks Vector Search. Batch dikirim jika sudah ada data deleteBatchSize yang siap. Contoh: 10.
deleteMaxBufferDuration	Opsional: Penundaan maksimum sebelum batch penghapusan dikirim ke Vector Search. Batch dikirim jika sudah ada data deleteBatchSize yang siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5 detik), Nm untuk menit (contoh: 12 menit), dan Nh untuk jam (contoh: 2 jam). Default: 10s.
dlqDirectory	Opsional: Jalur untuk menyimpan data yang belum diproses dengan alasan kegagalan tersebut untuk diproses. Defaultnya 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 perubahan. Nilai defaultnya adalah kosong.
bigtableChangeStreamMetadataTableTableId	Opsional: ID tabel metadata konektor perubahan stream Bigtable yang akan digunakan. Jika tidak disediakan, tabel metadata konektor aliran data perubahan Bigtable akan otomatis dibuat selama alur pipeline. Nilai defaultnya adalah kosong.
bigtableChangeStreamCharset	Opsional: Bigtable mengubah aliran nama charset saat membaca nilai dan penentu kolom. Defaultnya adalah UTF-8.
bigtableChangeStreamStartTimestamp	Opsional: DateTime awal, inklusif, yang akan digunakan untuk membaca aliran data perubahan (https://tools.ietf.org/html/rfc3339). Misalnya, 2022-05-05T07:59:59Z. Setelan defaultnya adalah stempel waktu saat pipeline dimulai.
bigtableChangeStreamIgnoreColumnFamilies	Opsional: Daftar perubahan nama keluarga kolom yang dipisahkan koma yang tidak akan ditangkap. Nilai defaultnya adalah kosong.
bigtableChangeStreamIgnoreColumns	Opsional: Daftar perubahan nama kolom yang dipisahkan koma yang tidak akan ditangkap. Nilai defaultnya adalah kosong.
bigtableChangeStreamName	Opsional: Nama unik untuk pipeline klien. Parameter ini memungkinkan Anda melanjutkan pemrosesan dari titik di mana pipeline yang sebelumnya berjalan telah berhenti. Setelan defaultnya adalah nama yang dibuat secara otomatis. Lihat log tugas Dataflow untuk nilai yang digunakan.
bigtableChangeStreamResume	Opsional: Jika disetel ke benar (true), pipeline baru akan melanjutkan pemrosesan dari titik di mana 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 jalur pipeline. Jika disetel ke salah (false), pipeline baru akan dimulai. Jika pipeline dengan nama yang sama seperti bigtableChangeStreamName sudah berjalan sebelumnya untuk sumber tertentu, pipeline baru akan gagal dimulai. Nilai defaultnya adalah false (salah).
bigtableReadProjectId	Opsional: Project yang akan dibaca data Bigtable. Default untuk parameter ini adalah project tempat pipeline Dataflow dijalankan.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Job name, masukkan nama tugas yang unik.
3. Opsional: Untuk Regional endpoint, pilih nilai dari menu drop-down. Region default 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: Agar dapat menggunakan Google Cloud CLI untuk menjalankan flex template, 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 pekerjaan 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 non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • gunakan nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan disarangkan dalam masing-masing folder induk bertanggal di bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi template terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• EMBEDDING_COLUMN: kolom Penyematan
• EMBEDDING_BYTE_SIZE: Ukuran byte dari array embedding. Bisa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Penelusuran Vektor
• 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 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 pekerjaan 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 non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • gunakan nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan disarangkan dalam masing-masing folder induk bertanggal di bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi template terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• EMBEDDING_COLUMN: kolom Penyematan
• EMBEDDING_BYTE_SIZE: Ukuran byte dari array embedding. Bisa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Penelusuran Vektor
• 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);
  }
}