이 페이지는 Cloud Translation API를 통해 번역되었습니다.

Pub/Sub to Java Database Connectivity(JDBC) 템플릿

Pub/Sub to JDBC(Java Database Connectivity) 템플릿은 기존 Pub/Sub 구독의 데이터를 JSON 문자열로 수집하고 결과 레코드를 JDBC에 쓰는 스트리밍 파이프라인입니다.

파이프라인 요구사항

파이프라인을 실행하기 전에 Pub/Sub 구독이 있어야 합니다.
파이프라인을 실행하기 전에 JDBC 소스가 있어야 합니다.
Pub/Sub 출력 데드 레터 주제는 파이프라인을 실행하기 전에 존재해야 합니다.

템플릿 매개변수

매개변수	설명
`driverClassName`	JDBC 드라이버 클래스 이름입니다. 예를 들면 `com.mysql.jdbc.Driver`입니다.
`connectionUrl`	JDBC 연결 URL 문자열입니다. 예를 들면 `jdbc:mysql://some-host:3306/sampledb`입니다. 이 값을 Cloud KMS 키로 암호화된 Base64 인코딩 문자열로 전달할 수 있습니다. Base64로 인코딩된 문자열에서 공백 문자를 삭제하세요.
`driverJars`	JDBC 드라이버의 쉼표로 구분된 Cloud Storage 경로입니다. 예를 들면 `gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar`입니다.
`username`	(선택사항) JDBC 연결에 사용할 사용자 이름입니다. 이 값을 Cloud KMS 키로 암호화된 Base64 인코딩된 문자열로 전달할 수 있습니다.
`password`	(선택사항) JDBC 연결에 사용할 비밀번호입니다. 이 값을 Cloud KMS 키로 암호화된 Base64 인코딩된 문자열로 전달할 수 있습니다.
`connectionProperties`	(선택사항) JDBC 연결에 사용할 속성 문자열입니다. 문자열 형식은 `[propertyName=property;]*`여야 합니다. 예를 들면 `unicode=true;characterEncoding=UTF-8`입니다.
`statement`	데이터베이스에 대해 실행할 문입니다. 이 문은 순서에 관계없이 테이블의 열 이름을 지정해야 합니다. 지정된 열 이름의 값만 JSON에서 읽고 문에 추가됩니다. 예를 들면 `INSERT INTO tableName (column1, column2) VALUES (?,?)`입니다.
`inputSubscription`	읽어올 Pub/Sub 입력 구독으로, `projects/<project>/subscriptions/<subscription>` 형식입니다.
`outputDeadletterTopic`	전달할 수 없는 메시지를 전달할 Pub/Sub 주제입니다. 예를 들면 `projects/<project-id>/topics/<topic-name>`입니다.
`KMSEncryptionKey`	(선택사항) 사용자 이름, 비밀번호, 연결 문자열을 복호화하는 Cloud KMS 암호화 키입니다. Cloud KMS 키가 전달되면 사용자 이름, 비밀번호, 연결 문자열이 모두 암호화되어 전달되어야 합니다.
`extraFilesToStage`	작업자에 스테이징할 파일의 쉼표로 구분된 Cloud Storage 경로 또는 Secret Manager 보안 비밀입니다. 이러한 파일은 각 작업자의 `/extra_files` 디렉터리에 저장됩니다. 예를 들면 `gs://<my-bucket>/file.txt,projects/<project-id>/secrets/<secret-id>/versions/<version-id>`입니다.

템플릿 실행

콘솔gcloudAPI

Dataflow 템플릿에서 작업 만들기 페이지로 이동합니다.

템플릿에서 작업 만들기로 이동

작업 이름 필드에 고유한 작업 이름을 입력합니다.
(선택사항): 리전 엔드포인트의 드롭다운 메뉴에서 값을 선택합니다. 기본 리전은 us-central1입니다.
Dataflow 작업을 실행할 수 있는 리전 목록은 Dataflow 위치를 참조하세요.
Dataflow 템플릿 드롭다운 메뉴에서 the Pub/Sub to JDBC template을 선택합니다.
제공된 매개변수 필드에 매개변수 값을 입력합니다.
작업 실행을 클릭합니다.

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location gs://dataflow-templates-REGION_NAME/VERSION/flex/PubSub_to_Jdbc \
    --region REGION_NAME \
    --parameters \
driverClassName=DRIVER_CLASS_NAME,\
connectionURL=JDBC_CONNECTION_URL,\
driverJars=DRIVER_PATHS,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
connectionProperties=CONNECTION_PROPERTIES,\
statement=SQL_STATEMENT,\
inputSubscription=INPUT_SUBSCRIPTION,\
outputDeadletterTopic=OUTPUT_DEADLETTER_TOPIC,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

다음을 바꿉니다.

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)
DRIVER_CLASS_NAME: 드라이버 클래스 이름입니다.
JDBC_CONNECTION_URL: JDBC 연결 URL
DRIVER_PATHS: JDBC 드라이버의 쉼표로 구분된 Cloud Storage 경로
CONNECTION_USERNAME: JDBC 연결 사용자 이름
CONNECTION_PASSWORD: JDBC 연결 비밀번호
CONNECTION_PROPERTIES: JDBC 연결 속성(필요한 경우)
SQL_STATEMENT: 데이터베이스에 대해 실행할 SQL 문
INPUT_SUBSCRIPTION: 읽어올 Pub/Sub 입력 구독
OUTPUT_DEADLETTER_TOPIC: 전달할 수 없는 메시지를 전달할 Pub/Sub
KMS_ENCRYPTION_KEY: Cloud KMS 암호화 키

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "JDBC_CONNECTION_URL",
       "driverJars": "DRIVER_PATHS",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "statement": "SQL_STATEMENT",
       "inputSubscription": "INPUT_SUBSCRIPTION",
       "outputDeadletterTopic": "OUTPUT_DEADLETTER_TOPIC",
       "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
   },
   "environment": { "zone": "us-central1-f" },
}

다음을 바꿉니다.

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)
DRIVER_CLASS_NAME: 드라이버 클래스 이름입니다.
JDBC_CONNECTION_URL: JDBC 연결 URL
DRIVER_PATHS: JDBC 드라이버의 쉼표로 구분된 Cloud Storage 경로
CONNECTION_USERNAME: JDBC 연결 사용자 이름
CONNECTION_PASSWORD: JDBC 연결 비밀번호
CONNECTION_PROPERTIES: JDBC 연결 속성(필요한 경우)
SQL_STATEMENT: 데이터베이스에 대해 실행할 SQL 문
INPUT_SUBSCRIPTION: 읽어올 Pub/Sub 입력 구독
OUTPUT_DEADLETTER_TOPIC: 전달할 수 없는 메시지를 전달할 Pub/Sub
KMS_ENCRYPTION_KEY: Cloud KMS 암호화 키

템플릿 소스 코드

Java

/*
 * Copyright (C) 2021 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 static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.io.DynamicJdbcIO;
import com.google.cloud.teleport.v2.options.PubsubToJdbcOptions;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.utils.JsonStringToQueryMapper;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Splitter;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link PubsubToJdbc} streaming pipeline reads data from Google Cloud PubSub and publishes to
 * JDBC.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Pubsub_to_Jdbc.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Pubsub_to_Jdbc",
    category = TemplateCategory.STREAMING,
    displayName = "Pub/Sub to JDBC",
    description =
        "The Pub/Sub to Java Database Connectivity (JDBC) template is a streaming pipeline that ingests data from a "
            + "pre-existing Cloud Pub/Sub subscription as JSON strings, and writes the resulting records to JDBC.",
    optionsClass = PubsubToJdbcOptions.class,
    flexContainerName = "pubsub-to-jdbc",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/pubsub-to-jdbc",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Pub/Sub subscription must exist prior to running the pipeline.",
      "The JDBC source must exist prior to running the pipeline.",
      "The Cloud Pub/Sub output deadletter topic must exist prior to running the pipeline.",
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class PubsubToJdbc {

  /* Logger for class.*/
  private static final Logger LOG = LoggerFactory.getLogger(PubsubToJdbc.class);

  /** String/String Coder for FailsafeElement. */
  public static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    PubsubToJdbcOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PubsubToJdbcOptions.class);

    run(options);
  }

  /**
   * Runs a pipeline which reads message from Pub/Sub and writes to JDBC.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(PubsubToJdbcOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    LOG.info("Starting Pubsub-to-Jdbc Pipeline.");

    /*
     * Steps:
     *  1) Read data from a Pub/Sub subscription
     *  2) Write to Jdbc Table
     *  3) Write errors to deadletter topic
     */
    PCollection<String> pubsubData =
        pipeline.apply(
            "readFromPubSubSubscription",
            PubsubIO.readStrings().fromSubscription(options.getInputSubscription()));

    DynamicJdbcIO.DynamicDataSourceConfiguration dataSourceConfiguration =
        DynamicJdbcIO.DynamicDataSourceConfiguration.create(
                options.getDriverClassName(),
                maybeDecrypt(options.getConnectionUrl(), options.getKMSEncryptionKey()).get())
            .withDriverJars(options.getDriverJars());
    if (options.getUsername() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withUsername(
              maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()).get());
    }
    if (options.getPassword() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withPassword(
              maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()).get());
    }
    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    PCollection<FailsafeElement<String, String>> errors =
        pubsubData
            .apply(
                "writeToJdbc",
                DynamicJdbcIO.<String>write()
                    .withDataSourceConfiguration(dataSourceConfiguration)
                    .withStatement(options.getStatement())
                    .withPreparedStatementSetter(
                        new JsonStringToQueryMapper(getKeyOrder(options.getStatement()))))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    errors.apply(
        "WriteFailedRecords",
        ErrorConverters.WriteStringMessageErrorsToPubSub.newBuilder()
            .setErrorRecordsTopic(options.getOutputDeadletterTopic())
            .build());

    return pipeline.run();
  }

  private static List<String> getKeyOrder(String statement) {
    int startIndex = statement.indexOf("(");
    int endIndex = statement.indexOf(")");
    String data = statement.substring(startIndex + 1, endIndex);
    return Splitter.on(',').trimResults().splitToList(data);
  }
}

다음 단계

Dataflow 템플릿 알아보기
Google 제공 템플릿 목록 참조