Cloud Spanner change streams to Pub/Sub 템플릿

Cloud Spanner change streams to Pub/Sub

Cloud Spanner change stream to Pub/Sub 템플릿은 Cloud Spanner 데이터 변경 레코드를 스트리밍하고 Dataflow Runner V2를 사용하여 Pub/Sub 주제에 쓰는 스트리밍 파이프라인입니다.

새 Pub/Sub 주제로 데이터를 출력하려면 먼저 주제를 만들어야 합니다. 만들어진 후에는 Pub/Sub에서 자동으로 구독을 생성하고 새 주제에 연결합니다. 데이터를 존재하지 않는 Pub/Sub 주제에 출력하려고 하면 Dataflow 파이프라인에서 예외가 발생하고 파이프라인이 연결을 계속 시도하므로 파이프라인이 중단됩니다.

필요한 Pub/Sub 주제가 이미 존재하면 데이터를 해당 주제로 출력할 수 있습니다.

자세한 내용은 변경 내역 정보, Dataflow를 사용하여 변경 내역 연결 빌드, 변경 내역 권장사항을 참조하세요.

파이프라인 요구사항

• 파이프라인을 실행하기 전에 Cloud Spanner 인스턴스가 있어야 합니다.
• 파이프라인을 실행하기 전에 Cloud Spanner 데이터베이스가 있어야 합니다.
• 파이프라인을 실행하기 전에 Cloud Spanner 메타데이터 인스턴스가 있어야 합니다.
• 파이프라인을 실행하기 전에 Cloud Spanner 메타데이터 데이터베이스가 있어야 합니다.
• 파이프라인을 실행하기 전에 Cloud Spanner 변경 내역이 있어야 합니다.
• 파이프라인을 실행하기 전에 Pub/Sub 주제가 있어야 합니다.

템플릿 매개변수

매개변수	설명
spannerInstanceId	변경 스트림을 읽어올 Cloud Spanner 인스턴스입니다.
spannerDatabase	변경 스트림을 읽어올 Cloud Spanner 데이터베이스입니다.
spannerMetadataInstanceId	변경 내역 커넥터 메타데이터 테이블에 사용할 Cloud Spanner 인스턴스입니다.
spannerMetadataDatabase	변경 내역 커넥터 메타데이터 테이블에 사용할 Cloud Spanner 데이터베이스입니다.
spannerChangeStreamName	읽어 올 Cloud Spanner 변경 스트림의 이름입니다.
pubsubTopic	변경 내역 출력을 위한 Pub/Sub 주제입니다.
spannerProjectId	(선택사항) 변경 내역을 읽어 올 프로젝트입니다. 변경 내역 커넥터 메타데이터 테이블이 생성되는 프로젝트이기도 합니다. 이 매개변수의 기본값은 Dataflow 파이프라인이 실행되는 프로젝트입니다.
spannerMetadataTableName	(선택사항) 사용할 Cloud Spanner 변경 내역 커넥터 메타데이터 테이블 이름입니다. 제공되지 않으면 Cloud Spanner는 파이프라인 흐름 변경 중에 스트림 커넥터 메타데이터 테이블을 자동으로 만듭니다. 기존 파이프라인을 업데이트할 때 이 매개변수를 제공해야 합니다. 다른 경우에는 이 매개변수를 사용하지 마세요.
rpcPriority	(선택사항) Cloud Spanner 호출의 요청 우선순위입니다. 값은 [HIGH,MEDIUM,LOW] 중 하나여야 합니다. (기본값: HIGH)
startTimestamp	(선택사항) 변경 내역을 읽는 데 사용할 시작 DateTime(경계 포함)입니다. 예를 들면 ex-2021-10-12T07:20:50.52Z입니다. 기본값은 파이프라인이 시작되는 시점의 타임스탬프, 즉 현재 시간입니다.
endTimestamp	(선택사항) 변경 내역을 읽는 데 사용할 종료 DateTime(경계 포함)입니다. 예를 들면 ex-2021-10-12T07:20:50.52Z입니다. 기본값은 미래의 무한대 시간입니다.
outputFileFormat	(선택사항) 출력 형식입니다. 출력은 여러 PubsubMessages로 래핑되고 Pub/Sub 주제로 전송됩니다. 허용되는 형식은 JSON 및 AVRO입니다. 기본값은 JSON입니다.
pubsubAPI	(선택사항)파이프라인을 구현하는 데 사용되는 Pub/Sub API입니다. 허용되는 API는 pubsubio 및 native_client입니다. 소수의 초당 쿼리 수(QPS)에서는 native_client 지연 시간이 더 짧습니다. 많은 수의 QPS에서 pubsubio가 더 우수하고 안정적인 성능을 제공합니다. 기본값은 pubsubio입니다.

템플릿 실행

콘솔

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

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

4. Dataflow 템플릿 드롭다운 메뉴에서 the Cloud Spanner change streams to Pub/Sub template을 선택합니다.
5. 제공된 매개변수 필드에 매개변수 값을 입력합니다.
6. 작업 실행을 클릭합니다.

gcloud

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

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

    gcloud beta dataflow flex-template run JOB_NAME \
        --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Spanner_Change_Streams_to_PubSub \
        --region REGION_NAME \
        --parameters \
    spannerInstanceId=SPANNER_INSTANCE_ID,\
    spannerDatabase=SPANNER_DATABASE,\
    spannerMetadataInstanceId=SPANNER_METADATA_INSTANCE_ID,\
    spannerMetadataDatabase=SPANNER_METADATA_DATABASE,\
    spannerChangeStreamName=SPANNER_CHANGE_STREAM,\
    pubsubTopic=PUBSUB_TOPIC
    

다음을 바꿉니다.

• JOB_NAME: 선택한 고유한 작업 이름
• VERSION: 사용할 템플릿 버전

  다음 값을 사용할 수 있습니다.

  • latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
  • 버전 이름(예: 2021-09-20-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
  주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
• REGION_NAME: Dataflow 작업을 배포할 리전 엔드포인트(예: us-central1)
• SPANNER_INSTANCE_ID: Cloud Spanner 인스턴스 ID
• SPANNER_DATABASE: Cloud Spanner 데이터베이스
• SPANNER_METADATA_INSTANCE_ID: Cloud Spanner 메타데이터 인스턴스 ID
• SPANNER_METADATA_DATABASE: Cloud Spanner 메타데이터 데이터베이스
• SPANNER_CHANGE_STREAM: Cloud Spanner 변경 내역
• PUBSUB_TOPIC: 변경 내역 출력을 위한 Pub/Sub 주제

API

REST API를 사용하여 템플릿을 실행하려면 HTTP POST 요청을 전송합니다. API 및 승인 범위에 대한 자세한 내용은 projects.templates.launch를 참조하세요.

  POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
  {
    "launch_parameter": {
        "jobName": "JOB_NAME",
        "parameters": {
            "spannerInstanceId": "SPANNER_INSTANCE_ID",
            "spannerDatabase": "SPANNER_DATABASE",
            "spannerMetadataInstanceId": "SPANNER_METADATA_INSTANCE_ID",
            "spannerMetadataDatabase": "SPANNER_METADATA_DATABASE",
            "spannerChangeStreamName": "SPANNER_CHANGE_STREAM",
            "pubsubTopic": "PUBSUB_TOPIC"
        },
        "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Spanner_Change_Streams_to_PubSub",
    }
  }
  

다음을 바꿉니다.

• PROJECT_ID: Dataflow 작업을 실행할 Cloud 프로젝트 ID
• JOB_NAME: 선택한 고유한 작업 이름
• VERSION: 사용할 템플릿 버전

  다음 값을 사용할 수 있습니다.

  • latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
  • 버전 이름(예: 2021-09-20-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
  주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
• LOCATION: Dataflow 작업을 배포할 리전 엔드포인트(예: us-central1)
• SPANNER_INSTANCE_ID: Cloud Spanner 인스턴스 ID
• SPANNER_DATABASE: Cloud Spanner 데이터베이스
• SPANNER_METADATA_INSTANCE_ID: Cloud Spanner 메타데이터 인스턴스 ID
• SPANNER_METADATA_DATABASE: Cloud Spanner 메타데이터 데이터베이스
• SPANNER_CHANGE_STREAM: Cloud Spanner 변경 내역
• PUBSUB_TOPIC: 변경 내역 출력을 위한 Pub/Sub 주제

템플릿 소스 코드

Java

GitHub에서 보기 의견 보내기

/*
 * Copyright (C) 2022 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;

import com.google.cloud.Timestamp;
import com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SpannerChangeStreamsToPubSubOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreamsToPubSub;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToPubSub} pipeline streams change stream record(s) and stores to
 * pubsub topic in user specified format. The sink data can be stored in a JSON Text or Avro data
 * format.
 */
@Template(
    name = "Spanner_Change_Streams_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Pub/Sub",
    description =
        "Streaming pipeline. Streams Spanner change stream data records and writes them into a"
            + " Pub/Sub topic using Dataflow Runner V2.",
    optionsClass = SpannerChangeStreamsToPubSubOptions.class,
    flexContainerName = "spanner-changestreams-to-pubsub",
    contactInformation = "https://cloud.google.com/support")
public class SpannerChangeStreamsToPubSub {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToPubSub.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    LOG.info("Starting Input Messages to Pub/Sub");

    SpannerChangeStreamsToPubSubOptions options =
        PipelineOptionsFactory.fromArgs(args).as(SpannerChangeStreamsToPubSubOptions.class);

    run(options);
  }

  private static String getProjectId(SpannerChangeStreamsToPubSubOptions options) {
    return options.getSpannerProjectId().isEmpty()
        ? options.getProject()
        : options.getSpannerProjectId();
  }

  public static PipelineResult run(SpannerChangeStreamsToPubSubOptions options) {
    LOG.info("Requested Message Format is " + options.getOutputDataFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);
    // Get the Spanner project, instance, database, metadata instance, metadata database
    // change stream, pubsub topic, and pubsub api parameters.
    String projectId = getProjectId(options);
    String instanceId = options.getSpannerInstanceId();
    String databaseId = options.getSpannerDatabase();
    String metadataInstanceId = options.getSpannerMetadataInstanceId();
    String metadataDatabaseId = options.getSpannerMetadataDatabase();
    String changeStreamName = options.getSpannerChangeStreamName();
    String pubsubTopicName = options.getPubsubTopic();
    String pubsubAPI = options.getPubsubAPI();

    // Retrieve and parse the start / end timestamps.
    Timestamp startTimestamp =
        options.getStartTimestamp().isEmpty()
            ? Timestamp.now()
            : Timestamp.parseTimestamp(options.getStartTimestamp());
    Timestamp endTimestamp =
        options.getEndTimestamp().isEmpty()
            ? Timestamp.MAX_VALUE
            : Timestamp.parseTimestamp(options.getEndTimestamp());

    // Add use_runner_v2 to the experiments option, since Change Streams connector is only supported
    // on Dataflow runner v2.
    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    if (!experiments.contains(USE_RUNNER_V2_EXPERIMENT)) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);

    String metadataTableName =
        options.getSpannerMetadataTableName() == null
            ? null
            : options.getSpannerMetadataTableName();

    final RpcPriority rpcPriority = options.getRpcPriority();

    final String errorMessage =
        "Invalid api:" + pubsubAPI + ". Supported apis: pubsubio, native_client";

    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(
                    SpannerConfig.create()
                        .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
                        .withProjectId(projectId)
                        .withInstanceId(instanceId)
                        .withDatabaseId(databaseId))
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName))
        .apply(
            "Convert each record to a PubsubMessage",
            FileFormatFactorySpannerChangeStreamsToPubSub.newBuilder()
                .setOutputDataFormat(options.getOutputDataFormat())
                .setProjectId(projectId)
                .setPubsubAPI(pubsubAPI)
                .setPubsubTopicName(pubsubTopicName)
                .build());
    return pipeline.run();
  }
}