Bigtable change streams to Vector Search 템플릿

이 템플릿은 스트리밍 파이프라인을 생성하여 Dataflow Runner V2를 사용해 Bigtable 데이터 변경 레코드를 스트리밍하고 Vertex AI 벡터 검색에 기록합니다.

파이프라인 요구사항

• Bigtable 소스 인스턴스가 있어야 합니다.
• Bigtable 소스 테이블이 있어야 하며 테이블에 변경 내역이 사용 설정되어 있어야 합니다.
• Bigtable 애플리케이션 프로필이 있어야 합니다.
• 벡터 검색 색인 경로가 있어야 합니다.

템플릿 매개변수

필수 매개변수

• embeddingColumn: 임베딩이 저장되는 정규화된 열 이름입니다. cf:col 형식입니다.
• embeddingByteSize: 임베딩 배열에 있는 각 항목의 바이트 크기입니다. 부동 소수점 수에는 4를, 더블에는 8을 사용합니다. 기본값은 4입니다.
• vectorSearchIndex: 변경사항을 스트리밍할 벡터 검색 색인의 형식은 'projects/{projectID}/locations/{region}/indexes/{indexID}'(선행 또는 후행 공백 없음)입니다. 예를 들어 projects/123/locations/us-east1/indexes/456입니다.
• bigtableChangeStreamAppProfile: Bigtable 애플리케이션 프로필 ID입니다. 애플리케이션 프로필에서 단일 클러스터 라우팅을 사용하고 단일 행 트랜잭션을 허용해야 합니다.
• bigtableReadInstanceId: 소스 Bigtable 인스턴스 ID입니다.
• bigtableReadTableId: 소스 Bigtable 테이블 ID입니다.

선택적 매개변수

• bigtableMetadataTableTableId: 메타데이터 테이블을 만드는 데 사용되는 테이블 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: 배치를 벡터 검색 색인에 삽입/업데이트(upsert)하기 전에 버퍼링할 최대 삽입/업데이트(upsert) 수입니다. upsertBatchSize 레코드가 준비되거나 레코드가 upsertBatchDelay 시간 동안 대기한 경우 배치가 전송됩니다. 예를 들면 10입니다. 기본값은 10입니다.
• upsertMaxBufferDuration: 삽입/업데이트(upsert) 배치가 벡터 검색으로 전송되기까지의 최대 지연 시간입니다. 삽입/업데이트(upsert) 배치 크기 레코드가 준비되었거나 레코드가 삽입/업데이트(upsert) 배치 지연 시간 동안 대기한 경우 배치가 전송됩니다. 허용되는 형식은 Ns(초, 예: 5s), Nm(분, 예: 12m), Nh(시, 예: 2h)입니다. 예를 들면 10s입니다. 기본값은 10s입니다.
• deleteMaxBatchSize: 벡터 검색 색인에서 일괄 삭제하기 전에 버퍼링할 최대 삭제 수입니다. deleteBatchSize 레코드가 준비되거나 레코드가 deleteBatchDelay 시간 동안 대기한 후 배치가 전송됩니다. 예를 들면 10입니다. 기본값은 10입니다.
• deleteMaxBufferDuration: 일괄 삭제가 벡터 검색에 전송되기까지의 최대 지연 시간입니다. 삭제가 준비된 레코드가 deleteBatchSize개에 도달하거나 레코드가 deleteBatchDelay 시간 동안 대기한 경우 배치가 전송됩니다. 허용되는 형식은 Ns(초, 예: 5s), Nm(분, 예: 12m), Nh(시, 예: 2h)입니다. 예를 들면 10s입니다. 기본값은 10s입니다.
• dlqDirectory: 메시지를 처리할 수 없는 이유와 함께 처리되지 않은 모든 레코드가 저장되는 경로입니다. 기본값은 Dataflow 작업의 임시 위치 아래에 있는 디렉터리입니다. 대부분의 상황에서는 기본값이면 충분합니다.
• bigtableChangeStreamMetadataInstanceId: Bigtable 변경 내역 메타데이터 인스턴스 ID입니다. 기본값은 빈 값입니다.
• bigtableChangeStreamMetadataTableTableId : Bigtable 변경 내역 커넥터 메타데이터 테이블의 ID입니다. 제공하지 않으면 파이프라인 실행 중에 Bigtable 변경 내역 커넥터 메타데이터 테이블이 자동으로 생성됩니다. 기본값은 빈 값입니다.
• bigtableChangeStreamCharset: Bigtable 변경 내역 문자 집합 이름입니다. 기본값은 UTF-8입니다.
• bigtableChangeStreamStartTimestamp: 변경 내역을 읽는 데 사용할 시작 타임스탬프(https://tools.ietf.org/html/rfc3339)입니다(경계 포함). 예를 들면 2022-05-05T07:59:59Z입니다. 기본값은 파이프라인 시작 시간의 타임스탬프입니다.
• bigtableChangeStreamIgnoreColumnFamilies: 무시할 column family 이름 변경 내역을 쉼표로 구분한 목록입니다. 기본값은 빈 값입니다.
• bigtableChangeStreamIgnoreColumns: 무시할 열 이름 변경 내역을 쉼표로 구분한 목록입니다. 예시: 'cf1:col1,cf2:col2' 기본값은 빈 값입니다.
• bigtableChangeStreamName: 클라이언트 파이프라인의 고유한 이름입니다. 이전에 실행 중이던 파이프라인이 중지된 지점에서 처리를 계속할 수 있습니다. 기본값은 자동으로 생성된 이름입니다. 사용된 값은 Dataflow 작업 로그를 참고하세요.
• bigtableChangeStreamResume: true로 설정하면 bigtableChangeStreamName 값이 동일하고, 이전에 실행 중이던 파이프라인이 중지된 지점부터 새 파이프라인이 처리를 계속합니다. 지정된 bigtableChangeStreamName 값의 파이프라인이 실행된 적이 없는 경우 새 파이프라인이 시작되지 않습니다. false로 설정하면 새 파이프라인이 시작됩니다. bigtableChangeStreamName 값이 동일한 파이프라인이 지정된 소스에 대해 이미 실행된 경우 새 파이프라인이 시작되지 않습니다. 기본값은 false입니다.
• bigtableReadChangeStreamTimeoutMs: Bigtable ReadChangeStream 요청 제한 시간(밀리초)입니다.
• bigtableReadProjectId: Bigtable 프로젝트 ID입니다. 기본값은 Dataflow 작업의 프로젝트입니다.

템플릿 실행

콘솔

1. Dataflow 템플릿에서 작업 만들기 페이지로 이동합니다.
템플릿에서 작업 만들기로 이동
2. 작업 이름 필드에 고유한 작업 이름을 입력합니다.
3. (선택사항) 리전 엔드포인트의 드롭다운 메뉴에서 값을 선택합니다. 기본 리전은 us-central1입니다.

   Dataflow 작업을 실행할 수 있는 리전 목록은 Dataflow 위치를 참조하세요.

4. Dataflow 템플릿 드롭다운 메뉴에서 the Bigtable Change Streams to Vector Search template을 선택합니다.
5. 제공된 파라미터 필드에 파라미터 값을 입력합니다.
6. 작업 실행을 클릭합니다.

gcloud CLI

참고: Google Cloud CLI를 사용하여 flex템플릿을 실행하려면 Google Cloud CLI 버전 284.0.0 이상이 필요합니다.

셸 또는 터미널에서 템플릿을 실행합니다.

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 애플리케이션 프로필 ID
• BIGTABLE_READ_INSTANCE_ID: 소스 Bigtable 인스턴스 ID
• BIGTABLE_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 프로젝트 ID
• JOB_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 애플리케이션 프로필 ID
• BIGTABLE_READ_INSTANCE_ID: 소스 Bigtable 인스턴스 ID
• BIGTABLE_READ_TABLE_ID: 소스 Bigtable 테이블 ID

템플릿 소스 코드

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