이 템플릿은 MongoDB 변경 내역에서 작동하는 스트리밍 파이프라인을 만듭니다. 이 템플릿을 사용하려면 변경 내역 데이터를 Pub/Sub에 게시합니다. 파이프라인은 Pub/Sub에서 JSON 레코드를 읽고 BigQuery에 씁니다. BigQuery에 기록되는 레코드의 형식은 MongoDB to BigQuery 일괄 템플릿과 동일합니다.
파이프라인 요구사항
대상 BigQuery 데이터 세트가 있어야 합니다.
Dataflow 작업자 머신에서 소스 MongoDB 인스턴스에 액세스할 수 있어야 합니다.
변경 내역을 읽으려면 Pub/Sub 주제를 만들어야 합니다.
파이프라인이 실행되는 동안 MongoDB 변경 내역에서 변경 데이터 캡처(CDC) 이벤트를 리슨하고 Pub/Sub에 JSON 레코드로 게시합니다. Pub/Sub에 메시지 게시에 대한 자세한 내용은 주제에 메시지 게시를 참조하세요.
템플릿 매개변수
필수 매개변수
mongoDbUri: mongodb+srv://:@ 형식의 MongoDB 연결 URI입니다.
database: 컬렉션을 읽을 MongoDB의 데이터베이스입니다. (예: my-db)
collection: MongoDB 데이터베이스 내부의 컬렉션 이름입니다. (예: my-collection)
userOption : 사용자 옵션: FLATTEN 또는 NONE. FLATTEN은 문서를 단일 수준으로 평면화합니다. NONE은 전체 문서를 JSON 문자열로 저장합니다. 기본값은 NONE입니다.
inputTopic : 'projects/your-project-id/topics/your-topic-name' 형식의 입력을 읽어올 Pub/Sub 주제입니다(예: projects/your-project-id/topics/your-topic-name).
outputTableSpec: 출력을 작성할 BigQuery 테이블 위치입니다. 이름은 <project>:<dataset>.<table_name> 형식이어야 합니다. 테이블 스키마가 입력 객체와 일치해야 합니다.
선택적 매개변수
useStorageWriteApiAtLeastOnce: 이 매개변수는 'BigQuery Storage Write API 사용'이 설정된 경우에만 적용됩니다. 사용 설정하면 Storage Write API에 최소 1회의 시맨틱스가 사용되고 그렇지 않은 경우 정확히 한 번의 시맨틱스가 사용됩니다. 기본값은 false입니다.
KMSEncryptionKey : mongodb URI 연결 문자열을 복호화하는 Cloud KMS 암호화 키입니다. Cloud KMS 키가 전달되면 mongodb URI 연결 문자열이 모두 암호화되어 전달되어야 합니다. (예: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key).
useStorageWriteApi: true이면 파이프라인은 BigQuery에 데이터를 쓸 때 Storage Write API를 사용합니다(https://cloud.google.com/blog/products/data-analytics/streaming-data-into-bigquery-using-storage-write-api 참조). 기본값은 false입니다. 정확히 한 번 모드에서 Storage Write API를 사용할 때는 'BigQuery Storage Write API의 스트림 수' 및 'BigQuery Storage Write API의 트리거 빈도(초)'와 같은 매개변수를 설정해야 합니다. Dataflow 적어도 한 번 모드를 사용 설정하거나 useStorageWriteApiAtLeastOnce 매개변수를 true로 설정하면 스트림 수나 트리거 빈도를 설정할 필요가 없습니다.
/*
* Copyright (C) 2019 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.mongodb.templates;
import static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.BigQueryWriteOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.JavascriptDocumentTransformerOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.MongoDbOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.PubSubOptions;
import com.google.cloud.teleport.v2.mongodb.templates.MongoDbToBigQueryCdc.Options;
import com.google.cloud.teleport.v2.options.BigQueryStorageApiStreamingOptions;
import com.google.cloud.teleport.v2.transforms.JavascriptDocumentTransformer.TransformDocumentViaJavascript;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import java.io.IOException;
import javax.script.ScriptException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The {@link MongoDbToBigQueryCdc} pipeline is a streaming pipeline which reads data pushed to
* PubSub from MongoDB Changestream and outputs the resulting records to BigQuery.
*
* <p>Check out <a
* href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/mongodb-to-googlecloud/README_MongoDB_to_BigQuery_CDC.md">README</a>
* for instructions on how to use or modify this template.
*/
@Template(
name = "MongoDB_to_BigQuery_CDC",
category = TemplateCategory.STREAMING,
displayName = "MongoDB to BigQuery (CDC)",
description =
"The MongoDB to BigQuery CDC (Change Data Capture) template is a streaming pipeline that works together with MongoDB change streams. "
+ "The pipeline reads the JSON records pushed to Pub/Sub via a MongoDB change stream and writes them to BigQuery as specified by the <code>userOption</code> parameter.",
optionsClass = Options.class,
flexContainerName = "mongodb-to-bigquery-cdc",
documentation =
"https://cloud.google.com/dataflow/docs/guides/templates/provided/mongodb-change-stream-to-bigquery",
contactInformation = "https://cloud.google.com/support",
preview = true,
requirements = {
"The target BigQuery dataset must exist.",
"The source MongoDB instance must be accessible from the Dataflow worker machines.",
"The change stream pushing changes from MongoDB to Pub/Sub should be running."
},
streaming = true,
supportsAtLeastOnce = true)
public class MongoDbToBigQueryCdc {
private static final Logger LOG = LoggerFactory.getLogger(MongoDbToBigQuery.class);
/** Options interface. */
public interface Options
extends PipelineOptions,
MongoDbOptions,
PubSubOptions,
BigQueryWriteOptions,
JavascriptDocumentTransformerOptions,
BigQueryStorageApiStreamingOptions {
// Hide the UseStorageWriteApiAtLeastOnce in the UI, because it will automatically be turned
// on when pipeline is running on ALO mode and using the Storage Write API
@TemplateParameter.Boolean(
order = 1,
optional = true,
parentName = "useStorageWriteApi",
parentTriggerValues = {"true"},
description = "Use at at-least-once semantics in BigQuery Storage Write API",
helpText =
"This parameter takes effect only if \"Use BigQuery Storage Write API\" is enabled. If"
+ " enabled the at-least-once semantics will be used for Storage Write API, otherwise"
+ " exactly-once semantics will be used.",
hiddenUi = true)
@Default.Boolean(false)
@Override
Boolean getUseStorageWriteApiAtLeastOnce();
void setUseStorageWriteApiAtLeastOnce(Boolean value);
}
/** class ParseAsDocumentsFn. */
private static class ParseAsDocumentsFn extends DoFn<String, Document> {
@ProcessElement
public void processElement(ProcessContext context) {
context.output(Document.parse(context.element()));
}
}
/**
* Main entry point for pipeline execution.
*
* @param args Command line arguments to the pipeline.
*/
public static void main(String[] args)
throws ScriptException, IOException, NoSuchMethodException {
UncaughtExceptionLogger.register();
Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
BigQueryIOUtils.validateBQStorageApiOptionsStreaming(options);
run(options);
}
/** Pipeline to read data from PubSub and write to MongoDB. */
public static boolean run(Options options)
throws ScriptException, IOException, NoSuchMethodException {
options.setStreaming(true);
Pipeline pipeline = Pipeline.create(options);
String userOption = options.getUserOption();
String inputOption = options.getInputTopic();
TableSchema bigquerySchema;
// Get MongoDbUri
String mongoDbUri = maybeDecrypt(options.getMongoDbUri(), options.getKMSEncryptionKey()).get();
if (options.getJavascriptDocumentTransformFunctionName() != null
&& options.getJavascriptDocumentTransformGcsPath() != null) {
bigquerySchema =
MongoDbUtils.getTableFieldSchemaForUDF(
mongoDbUri,
options.getDatabase(),
options.getCollection(),
options.getJavascriptDocumentTransformGcsPath(),
options.getJavascriptDocumentTransformFunctionName(),
options.getUserOption());
} else {
bigquerySchema =
MongoDbUtils.getTableFieldSchema(
mongoDbUri, options.getDatabase(), options.getCollection(), options.getUserOption());
}
pipeline
.apply("Read PubSub Messages", PubsubIO.readStrings().fromTopic(inputOption))
.apply(
"RTransform string to document",
ParDo.of(
new DoFn<String, Document>() {
@ProcessElement
public void process(ProcessContext c) {
Document document = Document.parse(c.element());
c.output(document);
}
}))
.apply(
"UDF",
TransformDocumentViaJavascript.newBuilder()
.setFileSystemPath(options.getJavascriptDocumentTransformGcsPath())
.setFunctionName(options.getJavascriptDocumentTransformFunctionName())
.build())
.apply(
"Read and transform data",
ParDo.of(
new DoFn<Document, TableRow>() {
@ProcessElement
public void process(ProcessContext c) {
Document document = c.element();
TableRow row = MongoDbUtils.getTableSchema(document, userOption);
c.output(row);
}
}))
.apply(
BigQueryIO.writeTableRows()
.to(options.getOutputTableSpec())
.withSchema(bigquerySchema)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
pipeline.run();
return true;
}
}
