Bigtable-Änderungsstreams zur Vorlage für die Vektorsuche

Diese Vorlage erstellt eine Streamingpipeline, um Bigtable-Datenänderungsdatensätze zu streamen und sie mit Dataflow Runner V2 in die Vertex AI Vektorsuche zu schreiben.

Pipelineanforderungen

• Die Bigtable-Quellinstanz muss vorhanden sein.
• Die Bigtable-Quelltabelle muss vorhanden sein und für die Tabelle müssen Änderungsstreams aktiviert sein.
• Das Bigtable-Anwendungsprofil muss vorhanden sein.
• Der Indexpfad für die Vektorsuche muss vorhanden sein.

Vorlagenparameter

Parameter	Beschreibung
embeddingColumn	Der voll qualifizierte Spaltenname, in dem die Einbettungen gespeichert sind. Im Format cf:col.
embeddingByteSize	Die Bytegröße jedes Eintrags im Einbettungsarray. Verwenden Sie 4 für Gleitkommazahl und 8 für Double. Die Standardeinstellung ist 4.
vectorSearchIndex	Der Vektorsuchindex, in den Änderungen gestreamt werden, im Format "projects/{projectID}/locations/{region}/indexes/{indexID}" (keine voran- oder nachgestellten Leerzeichen). Beispiel: projects/123/locations/us-east1/indexes/456.
bigtableChangeStreamAppProfile	Das Anwendungsprofil, das zur Unterscheidung von Arbeitslasten in Bigtable verwendet wird.
bigtableReadInstanceId	Die ID der Bigtable-Instanz, die die Tabelle enthält.
bigtableReadTableId	Die Bigtable-Tabelle, aus der gelesen werden soll.
bigtableMetadataTableTableId	Optional: ID für die erstellte Metadatentabelle. Wenn nichts festgelegt ist, generiert Bigtable eine ID.
crowdingTagColumn	Optional: Der voll qualifizierte Spaltenname, in dem das Crowding-Tag gespeichert ist, im Format cf:col.
allowRestrictsMappings	Optional: Die durch Kommas getrennten, vollständig qualifizierten Spaltennamen der Spalten, die als allow-Einschränkungen verwendet werden sollen, sowie ihre Aliasse. Jeder Spaltenname muss das Format cf:col->alias haben.
denyRestrictsMappings	Optional: Die durch Kommas getrennten, voll qualifizierten Spaltennamen der Spalten, die für deny-Einschränkungen verwendet werden sollen, mit ihren Aliassen. Jeder Spaltenname muss das Format cf:col->alias haben.
intNumericRestrictsMappings	Optional: Die durch Kommas getrennten, vollständig qualifizierten Spaltennamen der Spalten, die als Ganzzahl numeric_restricts verwendet werden sollen, mit ihren Aliassen. Jeder Spaltenname muss das Format cf:col->alias haben.
floatNumericRestrictsMappings	Optional: Die durch Kommas getrennten, vollständig qualifizierten Spaltennamen der Spalten, die als Gleitkommazahl (4 Byte) numeric_restricts verwendet werden sollen, mit ihren Aliassen. Jeder Spaltenname muss das Format cf:col->alias haben.
doubleNumericRestrictsMappings	Optional: Die durch Kommas getrennten, vollständig qualifizierten Spaltennamen der Spalten, die als Double (8 Byte) numeric_restricts verwendet werden sollen, mit ihren Aliassen. Jeder Spaltenname muss das Format cf:col->alias haben.
upsertMaxBatchSize	Optional: Die maximale Anzahl von Upserts zum Zwischenspeichern, bevor der Batch auf den Vektorsuchindex aktualisiert wird. Batches werden gesendet, wenn entweder upsertBatchSize-Datensätze bereit sind. Beispiel: 10.
upsertMaxBufferDuration	Optional: Die maximale Verzögerung, bevor ein Batch von Upserts an die Vektorsuche gesendet wird. Batches werden gesendet, wenn entweder upsertBatchSize-Datensätze bereit sind. Zulässige Formate sind: Ns für Sekunden (Beispiel: 5s), Nm für Minuten (Beispiel: 12m) und Nh für Stunden (Beispiel: 2h). Standard: 10s.
deleteMaxBatchSize	Optional: Die maximale Anzahl von Löschvorgängen, die zwischengespeichert werden sollen, bevor der Batch aus dem Vektorsuchindex gelöscht wird. Batches werden gesendet, wenn entweder deleteBatchSize-Datensätze bereit sind. Beispiel: 10
deleteMaxBufferDuration	Optional: Die maximale Verzögerung, bevor ein Batch von Löschvorgängen an die Vektorsuche gesendet wird. Batches werden gesendet, wenn entweder deleteBatchSize-Datensätze bereit sind. Zulässige Formate sind: Ns für Sekunden (Beispiel: 5s), Nm für Minuten (Beispiel: 12m) und Nh für Stunden (Beispiel: 2h). Standard: 10s.
dlqDirectory	Optional: Der Dateipfad zum Speichern nicht verarbeiteter Einträge mit einer Begründung für die fehlerhafte Verarbeitung. Der Standardwert ist ein Verzeichnis unter dem temporären Speicherort des Dataflow-Jobs. Der Standardwert ist für die meisten Szenarien geeignet.
bigtableChangeStreamMetadataInstanceId	Optional: Die Bigtable-Instanz, die für die Metadatentabelle des Connectors für Änderungsstreams verwendet werden soll. Die Standardeinstellung ist leer.
bigtableChangeStreamMetadataTableTableId	Optional: Die Bigtable-Änderung streamt die Metadatentabellen-ID des Connectors, die verwendet werden sollen. Wenn nicht angegeben, wird während des Pipelineablaufs automatisch eine Metadatentabelle für Bigtable-Änderungsstreams erstellt. Die Standardeinstellung ist leer.
bigtableChangeStreamCharset	Optional: Die Bigtable-Änderungsstreams codieren den Zeichensatznamen beim Lesen von Werten und Spaltenqualifizierern. Die Standardeinstellung ist UTF-8.
bigtableChangeStreamStartTimestamp	Optional: Die Start-DateTime (einschließlich), die zum Lesen von Änderungsstreams verwendet wird (https://tools.ietf.org/html/rfc3339). Beispiel: 2022-05-05T07:59:59Z Die Standardeinstellung ist der Zeitstempel für den Start der Pipeline.
bigtableChangeStreamIgnoreColumnFamilies	Optional: Eine durch Kommas getrennte Liste von Änderungen an den Namen von Spaltenfamilien, die nicht erfasst werden. Die Standardeinstellung ist leer.
bigtableChangeStreamIgnoreColumns	Optional: Eine durch Kommas getrennte Liste von Änderungen an Spaltennamen, die nicht erfasst werden. Die Standardeinstellung ist leer.
bigtableChangeStreamName	Optional: Ein eindeutiger Name für die Clientpipeline. Mit diesem Parameter wird die Verarbeitung ab dem Punkt fortgesetzt, an dem eine zuvor ausgeführte Pipeline beendet wurde. Standardmäßig wird ein automatisch generierter Name verwendet. Den verwendeten Wert finden Sie in den Dataflow-Joblogs.
bigtableChangeStreamResume	Optional: Wenn dieser Wert auf "true" gesetzt ist, setzt eine neue Pipeline die Verarbeitung ab dem Punkt fort, an dem eine zuvor ausgeführte Pipeline mit demselben Namen beendet wurde. Wenn eine Pipeline mit diesem Namen in der Vergangenheit noch nie ausgeführt wurde, kann die neue Pipeline nicht gestartet werden. Verwenden Sie den Parameter bigtableChangeStreamName, um die Pipelinelinie anzugeben. Wenn dieser Wert auf „false“ gesetzt ist, wird eine neue Pipeline gestartet. Wenn eine Pipeline mit demselben Namen wie bigtableChangeStreamName bereits in der Vergangenheit für die angegebene Quelle ausgeführt wurde, kann die neue Pipeline nicht starten kann. Die Standardeinstellung ist "false".
bigtableReadProjectId	Optional: Projekt, aus dem Bigtable-Daten gelesen werden sollen. Der Standardwert für diesen Parameter ist das Projekt, in dem die Dataflow-Pipeline ausgeführt wird.

Führen Sie die Vorlage aus.

Console

1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
Zur Seite "Job aus Vorlage erstellen“
2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
3. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

   Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the Bigtable Change Streams to Vector Search templateaus.
5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
6. Klicken Sie auf Job ausführen.

gcloud-CLI

Hinweis: Um flex-Vorlagen mit dem Google Cloud CLI auszuführen, benötigen Sie das Google Cloud CLI in der Version 284.0.0 oder höher.

Führen Sie die Vorlage in der Shell oder im Terminal aus:

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

Ersetzen Sie dabei Folgendes:

• JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• EMBEDDING_COLUMN: die Spalte „Embedding“
• EMBEDDING_BYTE_SIZE: die Bytegröße des Einbettungsarrays. Kann 4 oder 8 sein.
• VECTOR_SEARCH_INDEX: der Indexpfad der Vektorsuche
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: die Bigtable-Anwendungsprofil-ID
• BIGTABLE_READ_INSTANCE_ID: die Bigtable-Quellinstanz-ID
• BIGTABLE_READ_TABLE_ID: die Bigtable-Quelltabellen-ID.

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter 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" }
  }
}

Ersetzen Sie dabei Folgendes:

• PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
• JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• EMBEDDING_COLUMN: die Spalte „Embedding“
• EMBEDDING_BYTE_SIZE: die Bytegröße des Einbettungsarrays. Kann 4 oder 8 sein.
• VECTOR_SEARCH_INDEX: der Indexpfad der Vektorsuche
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: die Bigtable-Anwendungsprofil-ID
• BIGTABLE_READ_INSTANCE_ID: die Bigtable-Quellinstanz-ID
• BIGTABLE_READ_TABLE_ID: die Bigtable-Quelltabellen-ID

Quellcode der Vorlage

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);
  }
}