Aliran data perubahan Spanner ke template Cloud Storage

Template streaming perubahan Spanner ke Cloud Storage adalah pipeline streaming yang melakukan streaming rekaman perubahan data Spanner dan menuliskannya ke bucket Cloud Storage menggunakan Dataflow Runner v2.

Catatan: Konektor aliran perubahan Spanner tidak mendukung penghentian tugas. Untuk menghentikan pipeline, batalkan tugas alih-alih mengurasnya. Menguras tugas dengan konektor aliran perubahan Spanner dapat menyebabkan tugas tersebut macet dalam status pengurasan. Untuk mengetahui informasi selengkapnya, lihat Pengurasan dalam dokumentasi Spanner.

Pipeline mengelompokkan rekaman aliran perubahan Spanner ke dalam jendela berdasarkan stempel waktunya, dengan setiap jendela merepresentasikan durasi waktu yang panjangnya dapat Anda konfigurasi dengan template ini. Semua rekaman dengan stempel waktu yang termasuk dalam periode dijamin berada dalam periode tersebut; tidak ada kedatangan yang terlambat. Anda juga dapat menentukan sejumlah shard output; pipeline membuat satu file output Cloud Storage per jendela per shard. Dalam file output, rekaman tidak berurutan. File output dapat ditulis dalam format JSON atau AVRO, bergantung pada konfigurasi pengguna.

Perhatikan bahwa Anda dapat meminimalkan latensi jaringan dan biaya transportasi jaringan dengan menjalankan tugas Dataflow dari region yang sama dengan instance Spanner atau bucket Cloud Storage Anda. Jika Anda menggunakan sumber, tujuan, lokasi file staging, atau lokasi file sementara yang berada di luar region pekerjaan Anda, data Anda mungkin dikirim ke berbagai region. Lihat informasi selengkapnya tentang Region Dataflow.

Pelajari lebih lanjut aliran perubahan, cara membuat pipeline Dataflow aliran perubahan, dan praktik terbaik.

Persyaratan pipeline

• Instance Spanner harus ada sebelum menjalankan pipeline.
• Database Spanner harus ada sebelum menjalankan pipeline.
• Instance metadata Spanner harus ada sebelum menjalankan pipeline.
• Database metadata Spanner harus ada sebelum menjalankan pipeline.
• Aliran perubahan Spanner harus ada sebelum menjalankan pipeline.
• Bucket output Cloud Storage harus ada sebelum menjalankan pipeline.

Parameter template

Parameter yang diperlukan

• spannerInstanceId: ID instance Spanner untuk membaca data aliran perubahan.
• spannerDatabase: Database Spanner untuk membaca data aliran perubahan.
• spannerMetadataInstanceId: ID instance Spanner yang akan digunakan untuk tabel metadata konektor aliran perubahan.
• spannerMetadataDatabase: Database Spanner yang akan digunakan untuk tabel metadata konektor aliran perubahan.
• spannerChangeStreamName: Nama aliran data perubahan Spanner yang akan dibaca.
• gcsOutputDirectory: Jalur dan awalan nama file untuk menulis file output. Harus diakhiri dengan garis miring. Pemformatan DateTime digunakan untuk mengurai jalur direktori untuk pemformat tanggal & waktu. Contoh, gs://your-bucket/your-path.

Parameter opsional

• spannerProjectId: ID project Google Cloud yang berisi database Spanner untuk membaca aliran perubahan. Project ini juga merupakan tempat tabel metadata konektor aliran perubahan dibuat. Nilai default untuk parameter ini adalah project tempat pipeline Dataflow berjalan.
• spannerDatabaseRole: Peran database Spanner yang akan digunakan saat menjalankan template. Parameter ini hanya diperlukan jika pokok IAM yang menjalankan template adalah pengguna kontrol akses terperinci. Peran database harus memiliki hak istimewa SELECT pada aliran perubahan dan hak istimewa EXECUTE pada fungsi baca aliran perubahan. Untuk mengetahui informasi selengkapnya, lihat Kontrol akses terperinci untuk aliran perubahan (https://cloud.google.com/spanner/docs/fgac-change-streams).
• spannerMetadataTableName: Nama tabel metadata konektor aliran perubahan Spanner yang akan digunakan. Jika tidak disediakan, tabel metadata aliran perubahan Spanner akan dibuat secara otomatis selama eksekusi pipeline. Anda harus memberikan nilai untuk parameter ini saat memperbarui pipeline yang ada. Jika tidak, jangan gunakan parameter ini.
• startTimestamp: DateTime awal, inklusif, yang akan digunakan untuk membaca aliran perubahan, dalam format Ex-2021-10-12T07:20:50.52Z. Secara default, nilai ini adalah stempel waktu saat pipeline dimulai, yaitu waktu saat ini.
• endTimestamp: DateTime akhir, inklusif, yang akan digunakan untuk membaca aliran perubahan. Contoh, Ex-2021-10-12T07:20:50.52Z. Defaultnya adalah waktu tak terbatas di masa mendatang.
• spannerHost: Endpoint Cloud Spanner yang akan dipanggil dalam template. Hanya digunakan untuk pengujian. Contoh, https://spanner.googleapis.com. Nilai defaultnya adalah: https://spanner.googleapis.com.
• outputFileFormat: Format file Cloud Storage output. Format yang diizinkan adalah TEXT dan AVRO. Setelan defaultnya adalah AVRO.
• windowDuration: Durasi jendela adalah interval saat data ditulis ke direktori output. Konfigurasi durasi berdasarkan throughput pipeline. Misalnya, throughput yang lebih tinggi mungkin memerlukan ukuran jendela yang lebih kecil agar data sesuai dengan memori. Default-nya adalah 5m (lima menit), dengan minimum 1s (satu detik). Format yang diizinkan adalah: [int]s (untuk detik, contoh: 5s), [int]m (untuk menit, contoh: 12m), [int]h (untuk jam, contoh: 2h). Contoh, 5m.
• rpcPriority: Prioritas permintaan untuk panggilan Spanner. Nilainya harus berupa HIGH, MEDIUM, atau LOW. Nilai defaultnya adalah HIGH.
• outputFilenamePrefix: Awalan yang akan ditempatkan pada setiap file berwindow. Contoh, output-. Nilai defaultnya: output.
• numShards: Jumlah maksimum shard output yang dihasilkan saat menulis. Jumlah shard yang lebih tinggi berarti throughput yang lebih tinggi untuk menulis ke Cloud Storage, tetapi berpotensi menimbulkan biaya agregasi data yang lebih tinggi di seluruh shard saat memproses file Cloud Storage output. Nilai defaultnya adalah: 20.

Menjalankan template

Konsol

1. Buka halaman Dataflow Membuat tugas dari template.
Buka Membuat tugas dari template
2. Di kolom Nama tugas, masukkan nama tugas yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Template Dataflow, pilih the Cloud Spanner change streams to Google Cloud Storage template.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Jalankan tugas.

gcloud

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

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

Ganti kode berikut:

• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket—gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• SPANNER_INSTANCE_ID: ID instance Cloud Spanner
• SPANNER_DATABASE: Database Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID instance metadata Cloud Spanner
• SPANNER_METADATA_DATABASE: Database metadata Cloud Spanner
• SPANNER_CHANGE_STREAM: Aliran data perubahan Cloud Spanner
• GCS_OUTPUT_DIRECTORY: Lokasi file untuk output aliran perubahan

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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",
          "gcsOutputDirectory": "GCS_OUTPUT_DIRECTORY"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Spanner_Change_Streams_to_Google_Cloud_Storage",
   }
}

Ganti kode berikut:

• PROJECT_ID: Google Cloud Project ID tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama tugas unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket—gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• SPANNER_INSTANCE_ID: ID instance Cloud Spanner
• SPANNER_DATABASE: Database Cloud Spanner
• SPANNER_METADATA_INSTANCE_ID: ID instance metadata Cloud Spanner
• SPANNER_METADATA_DATABASE: Database metadata Cloud Spanner
• SPANNER_CHANGE_STREAM: Aliran data perubahan Cloud Spanner
• GCS_OUTPUT_DIRECTORY: Lokasi file untuk output aliran perubahan

Kode sumber template

Java

/*
 * 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.SpannerChangeStreamsToGcsOptions;
import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreams;
import com.google.cloud.teleport.v2.utils.DurationUtils;
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.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link SpannerChangeStreamsToGcs} pipeline streams change stream record(s) and stores to
 * Google Cloud Storage bucket in user specified format. The sink data can be stored in a Text or
 * Avro file format.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Spanner_Change_Streams_to_Google_Cloud_Storage.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Spanner_Change_Streams_to_Google_Cloud_Storage",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Spanner change streams to Cloud Storage",
    description = {
      "The Cloud Spanner change streams to Cloud Storage template is a streaming pipeline that streams Spanner data change records and writes them into a Cloud Storage bucket using Dataflow Runner V2.\n",
      "The pipeline groups Spanner change stream records into windows based on their timestamp, with each window representing a time duration whose length you can configure with this template. "
          + "All records with timestamps belonging to the window are guaranteed to be in the window; there can be no late arrivals. "
          + "You can also define a number of output shards; the pipeline creates one Cloud Storage output file per window per shard. "
          + "Within an output file, records are unordered. Output files can be written in either JSON or AVRO format, depending on the user configuration.\n",
      "Note that you can minimize network latency and network transport costs by running the Dataflow job from the same region as your Cloud Spanner instance or Cloud Storage bucket. "
          + "If you use sources, sinks, staging file locations, or temporary file locations that are located outside of your job's region, your data might be sent across regions. "
          + "See more about <a href=\"https://cloud.google.com/dataflow/docs/concepts/regional-endpoints\">Dataflow regional endpoints</a>.\n",
      "Learn more about <a href=\"https://cloud.google.com/spanner/docs/change-streams\">change streams</a>, <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow\">how to build change streams Dataflow pipelines</a>, and <a href=\"https://cloud.google.com/spanner/docs/change-streams/use-dataflow#best_practices\">best practices</a>."
    },
    optionsClass = SpannerChangeStreamsToGcsOptions.class,
    flexContainerName = "spanner-changestreams-to-gcs",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-spanner-change-streams-to-cloud-storage",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The Cloud Spanner instance must exist prior to running the pipeline.",
      "The Cloud Spanner database must exist prior to running the pipeline.",
      "The Cloud Spanner metadata instance must exist prior to running the pipeline.",
      "The Cloud Spanner metadata database must exist prior to running the pipeline.",
      "The Cloud Spanner change stream must exist prior to running the pipeline.",
      "The Cloud Storage output bucket must exist prior to running the pipeline."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class SpannerChangeStreamsToGcs {
  private static final Logger LOG = LoggerFactory.getLogger(SpannerChangeStreamsToGcs.class);
  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

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

    LOG.info("Starting Input Files to GCS");

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

    run(options);
  }

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

  public static PipelineResult run(SpannerChangeStreamsToGcsOptions options) {
    LOG.info("Requested File Format is " + options.getOutputFileFormat());
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    final Pipeline pipeline = Pipeline.create(options);

    // Get the Spanner project, instance, database, and change stream 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();

    // 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();
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withProjectId(projectId)
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId);
    // Propagate database role for fine-grained access control on change stream.
    if (options.getSpannerDatabaseRole() != null) {
      LOG.info("Setting database role on SpannerConfig: " + options.getSpannerDatabaseRole());
      spannerConfig =
          spannerConfig.withDatabaseRole(
              ValueProvider.StaticValueProvider.of(options.getSpannerDatabaseRole()));
    }
    LOG.info("Created SpannerConfig: " + spannerConfig);
    pipeline
        .apply(
            SpannerIO.readChangeStream()
                .withSpannerConfig(spannerConfig)
                .withMetadataInstance(metadataInstanceId)
                .withMetadataDatabase(metadataDatabaseId)
                .withChangeStreamName(changeStreamName)
                .withInclusiveStartAt(startTimestamp)
                .withInclusiveEndAt(endTimestamp)
                .withRpcPriority(rpcPriority)
                .withMetadataTable(metadataTableName))
        .apply(
            "Creating " + options.getWindowDuration() + " Window",
            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))
        .apply(
            "Write To GCS",
            FileFormatFactorySpannerChangeStreams.newBuilder().setOptions(options).build());

    return pipeline.run();
  }
}

Langkah berikutnya

• Pelajari template Dataflow.
• Lihat daftar template yang disediakan Google.