Template Sourcedb ke Spanner

Beta

Fitur ini tunduk pada "Persyaratan Penawaran Pra-GA" di bagian Persyaratan Layanan Umum dari Persyaratan Khusus Layanan. Fitur pra-GA tersedia "sebagaimana adanya" dan mungkin memiliki dukungan terbatas. Untuk mengetahui informasi selengkapnya, lihat deskripsi tahap peluncuran.

Template SourceDB ke Spanner adalah pipeline batch yang menyalin data dari database relasional ke database Spanner yang ada. Pipeline ini menggunakan JDBC untuk terhubung ke database relasional. Anda dapat menggunakan template ini untuk menyalin data dari database relasional mana pun dengan driver JDBC yang tersedia ke Spanner. Hanya mendukung jenis MySQL yang terbatas

Untuk lapisan perlindungan tambahan, Anda juga dapat meneruskan kunci Cloud KMS bersama dengan parameter nama pengguna, sandi, dan string koneksi yang dienkode Base64 dan dienkripsi dengan kunci Cloud KMS. Lihat endpoint enkripsi API Cloud KMS untuk mengetahui detail tambahan tentang mengenkripsi parameter nama pengguna, sandi, dan string koneksi Anda.

Catatan: Jika Anda berencana memvalidasi migrasi ini menggunakan template Validasi Data file Avro Cloud Storage ke Spanner, Anda harus mengisi parameter gcsOutputDirectory untuk membuat file Avro yang diperlukan.

Persyaratan pipeline

• Driver JDBC untuk database relasional harus tersedia.
• Tabel Spanner harus ada sebelum eksekusi pipeline.
• Tabel Spanner harus memiliki skema yang kompatibel.
• Database relasional harus dapat diakses dari subnet tempat Dataflow berjalan.

Parameter template

Parameter yang diperlukan

• sourceConfigURL: URL file konfigurasi koneksi sumber. Format file bergantung pada jenis sumber. Untuk Astra, file ini akan mengarah ke file konfigurasi koneksi Astra (contoh). Untuk JDBC, parameter ini akan mengarah ke file konfigurasi sharding JDBC (sample). Untuk Cassandra, kolom ini akan mengarah ke file konfigurasi driver Cassandra (sample). Parameter ini wajib diisi. Sebagai contoh, gs://your-bucket/source-config.json Nilai defaultnya adalah kosong.
• instanceId: Instance Cloud Spanner tujuan.
• databaseId: Database Cloud Spanner tujuan.
• projectId: Ini adalah nama project Cloud Spanner.
• outputDirectory: Direktori ini digunakan untuk membuang data yang gagal/dilewati/difilter dalam migrasi.

Parameter opsional

• sourceDbDialect: Nilai yang mungkin adalah CASSANDRA, MYSQL, POSTGRESQL, ORACLE, dan SQLSERVER. Nilai defaultnya: MYSQL.
• jdbcDriverJars: Daftar file JAR driver yang dipisahkan koma. Contoh, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar. Nilai defaultnya adalah kosong.
• jdbcDriverClassName: Nama class driver JDBC. Contoh, com.mysql.jdbc.Driver. Nilai defaultnya adalah: com.mysql.jdbc.Driver.
• tables: Tabel yang akan dimigrasikan dari sumber. Nilai defaultnya adalah kosong.
• numPartitions: Jumlah partisi. Ini, bersama dengan batas bawah dan atas, membentuk langkah partisi untuk ekspresi klausa WHERE yang dihasilkan dan digunakan untuk membagi kolom partisi secara merata. Jika input kurang dari 1, angka akan ditetapkan ke 1. Nilai defaultnya adalah: 0.
• fetchSize: Jumlah baris yang akan diambil per halaman yang dibaca untuk sumber JDBC. Jika tidak disetel, nilai ini akan disimpulkan secara otomatis dari jenis mesin pekerja dan perkiraan ukuran baris, yang akan kembali ke 50.000 baris jika tidak dapat disimpulkan (misalnya, saat jenis mesin pekerja tidak ditentukan). Jika dialek sumber adalah Mysql, lihat catatan di bawah. Pada akhirnya, hal ini diterjemahkan menjadi panggilan Statement.setFetchSize di lapisan Jdbc. Parameter ini HANYA boleh digunakan jika nilai default memunculkan error memori.Catatan untuk Sumber MySql: FetchSize diabaikan oleh konektor Mysql kecuali jika useCursorFetch=true juga merupakan bagian dari properti koneksi. Untuk dialek MySql, pipeline akan menambahkan useCursorFetch=true ke properti koneksi secara default, kecuali jika parameter fetchSize ditetapkan secara eksplisit ke 0.
• spannerHost: Endpoint Cloud Spanner yang akan dipanggil dalam template. Contoh, https://batch-spanner.googleapis.com. Secara default: https://batch-spanner.googleapis.com.
• maxConnections: Mengonfigurasi kumpulan koneksi JDBC di setiap pekerja dengan jumlah koneksi maksimum. Gunakan angka negatif untuk tanpa batas. Contoh, -1. Nilai defaultnya adalah: 0.
• sessionFilePath: Jalur file sesi di Cloud Storage yang berisi informasi pemetaan dari Spanner Migration Tool. Nilai defaultnya adalah kosong.
• transformationJarPath: Lokasi jar kustom di Cloud Storage yang berisi logika transformasi kustom untuk memproses data. Nilai defaultnya adalah kosong.
• transformationClassName: Nama class yang sepenuhnya memenuhi syarat yang memiliki logika transformasi kustom. Kolom ini wajib diisi jika transformationJarPath ditentukan. Nilai defaultnya adalah kosong.
• transformationCustomParameters: String yang berisi parameter kustom yang akan diteruskan ke class transformasi kustom. Nilai defaultnya adalah kosong.
• insertOnlyModeForSpannerMutations: Secara default, pipeline menggunakan Upsert untuk menulis baris ke Spanner. Artinya, baris yang ada akan ditimpa. Jika mode InsertOnly diaktifkan, penyisipan akan digunakan, bukan upsert, dan baris yang ada tidak akan ditimpa.
• batchSizeForSpannerMutations: BatchSize dalam byte untuk Mutasi Spanner. Jika ditetapkan kurang dari 0, default SpannerIO Apache Beam akan digunakan, yaitu 1 MB. Setel ini ke 0 atau 10, untuk menonaktifkan mutasi batch.
• spannerPriority: Prioritas permintaan untuk panggilan Cloud Spanner. Nilai harus berupa salah satu dari: [HIGH,MEDIUM,LOW]. Defaultnya adalah HIGH.
• tableOverrides: Ini adalah penggantian nama tabel dari sumber ke Spanner. Data tersebut ditulis dalam format berikut: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]Contoh ini menunjukkan pemetaan tabel Singers ke Vocalists dan tabel Albums ke Records. Contoh, [{Singers, Vocalists}, {Albums, Records}]. Nilai defaultnya adalah kosong.
• columnOverrides: Ini adalah penggantian nama kolom dari sumber ke spanner. Ditulis dalam format berikut: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]Perhatikan bahwa SourceTableName harus tetap sama dalam pasangan sumber dan Spanner. Untuk mengganti nama tabel, gunakan tableOverrides.Contoh ini menunjukkan pemetaan SingerName ke TalentName dan AlbumName ke RecordName dalam tabel Singers dan Albums. Contoh, [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]. Nilai defaultnya adalah kosong.
• schemaOverridesFilePath: File yang menentukan penggantian nama tabel dan kolom dari sumber ke Spanner. Nilai defaultnya adalah kosong.
• uniformizationStageCountHint: Petunjuk untuk jumlah tahap penyeragaman. Saat ini Hanya berlaku untuk sumber berbasis jdbc seperti MySQL atau PostgreSQL. Biarkan 0 atau default untuk menonaktifkan penyeragaman. Setel ke -1 untuk jumlah tahap log(numPartition). Jika ruang kunci utama sumber Anda didistribusikan secara merata (misalnya, kunci yang bertambah otomatis dengan banyak celah), sebaiknya biarkan tidak diaktifkan. Jika keyspace Anda tidak seragam, Anda mungkin akan menemukan VM yang lambat dalam eksekusi alur data. Dalam kasus seperti ini, Anda dapat menyetelnya ke -1 untuk mengaktifkan penyeragaman. Menetapkannya secara manual ke nilai selain 0 atau -1 akan membantu Anda menyetel kompromi antara overhead yang ditambahkan oleh tahap penyeragaman dan peningkatan performa karena distribusi tugas yang lebih baik.
• failureInjectionParameter: Parameter injeksi kegagalan. Hanya digunakan untuk pengujian. Nilai defaultnya adalah kosong.
• maxCommitDelay: Waktu jeda commit maksimum untuk mengoptimalkan throughput tulis di Spanner. Referensi https://cloud.google.com/spanner/docs/throughput-optimized-writes.Set -1 agar Spanner memilih nilai default. Tetapkan ke nilai positif untuk mengganti pertukaran throughput vs. latensi yang paling sesuai. Nilai defaultnya adalah -1.
• gcsOutputDirectory: Direktori ini digunakan untuk menulis file AVRO dari data yang dibaca dari sumber. Contoh, gs://your-bucket/your-path. Nilai defaultnya adalah kosong.
• disabledAlgorithms: Algoritma yang dipisahkan koma yang akan dinonaktifkan. Jika nilai ini disetel ke none, tidak ada algoritma yang dinonaktifkan. Gunakan parameter ini dengan hati-hati, karena algoritma yang dinonaktifkan secara default mungkin memiliki kerentanan atau masalah performa. Contoh, SSLv3, RC4.
• extraFilesToStage: Jalur Cloud Storage atau secret Secret Manager yang dipisahkan koma untuk file yang akan di-stage di worker. File ini disimpan di direktori /extra_files di setiap worker. Misalnya, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

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 template Sourcedb to Spanner.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Jalankan tugas.

gcloud CLI

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/VERSION/flex/Sourcedb_to_Spanner_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       sourceConfigURL=SOURCE_CONFIG_URL,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       projectId=PROJECT_ID,\
       outputDirectory=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/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/
  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
• SOURCE_CONFIG_URL: URL untuk terhubung ke host database sumber. Nilainya bisa 1. URL koneksi JDBC - yang harus berisi host, port, dan nama db sumber, serta dapat secara opsional berisi properti seperti autoReconnect, maxReconnects, dll. Format: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Jalur konfigurasi shard
• INSTANCE_ID: ID Instance Cloud Spanner.
• DATABASE_ID: ID Database Cloud Spanner.
• PROJECT_ID: Project ID Cloud Spanner.
• OUTPUT_DIRECTORY: Direktori Output untuk peristiwa yang gagal/dilewati/difilter

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
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "sourceConfigURL": "SOURCE_CONFIG_URL",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "projectId": "PROJECT_ID",
       "outputDirectory": "OUTPUT_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

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/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/
  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
• SOURCE_CONFIG_URL: URL untuk terhubung ke host database sumber. Nilainya bisa 1. URL koneksi JDBC - yang harus berisi host, port, dan nama db sumber, serta dapat secara opsional berisi properti seperti autoReconnect, maxReconnects, dll. Format: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Jalur konfigurasi shard
• INSTANCE_ID: ID Instance Cloud Spanner.
• DATABASE_ID: ID Database Cloud Spanner.
• PROJECT_ID: Project ID Cloud Spanner.
• OUTPUT_DIRECTORY: Direktori Output untuk peristiwa yang gagal/dilewati/difilter

Kode sumber template

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;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.source.ISrcToSpSourceConnector;
import com.google.cloud.teleport.v2.source.SourceConnectorFactory;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig;
import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.Optional;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
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.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.commons.lang3.StringUtils;

/**
 * A template that copies data from a relational database using JDBC to an existing Spanner
 * database.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Sourcedb_to_Spanner_Flex",
    category = TemplateCategory.BATCH,
    displayName = "Sourcedb to Spanner",
    description = {
      "The SourceDB to Spanner template is a batch pipeline that copies data from a relational"
          + " database into an existing Spanner database. This pipeline uses JDBC to connect to"
          + " the relational database. You can use this template to copy data from any relational"
          + " database with available JDBC drivers into Spanner. This currently only supports a limited set of types of MySQL",
    },
    optionsClass = SourceDbToSpannerOptions.class,
    flexContainerName = "source-db-to-spanner",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/sourcedb-to-spanner",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class SourceDbToSpanner {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link SourceDbToSpanner#run} method to start the
   * pipeline and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    SourceDbToSpannerOptions options = getSourceDbToSpannerOptions(args);
    run(options);
  }

  @VisibleForTesting
  protected static SourceDbToSpannerOptions getSourceDbToSpannerOptions(String[] args) {
    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    // Stage SSL certificates to extraFiles if required as per the pipeline options.
    // Ref https://cloud.google.com/dataflow/docs/guides/templates/ssl-certificates
    new CommonTemplateJvmInitializer().beforeProcessing(options);
    return options;
  }

  /**
   * Validates the provided pipeline options. TODO: move this to source connector.
   *
   * @param options The execution parameters to the pipeline.
   * @param sourceConnectionConfig Parsed source connection config.
   * @throws IllegalArgumentException if the provided options are invalid for the pipeline.
   */
  @VisibleForTesting
  static void validateOptions(
      SourceDbToSpannerOptions options, SourceConnectionConfig sourceConnectionConfig) {
    if (SourceDbToSpannerOptions.PG_SOURCE_DIALECT.equals(options.getSourceDbDialect())) {
      Preconditions.checkArgument(
          (sourceConnectionConfig instanceof JdbcShardConfig),
          "Postgresql dialect should have JDBC source config.");
      for (Shard shard : ((JdbcShardConfig) sourceConnectionConfig).getShardConfigs()) {
        if (StringUtils.isNotBlank(shard.getNamespace())
            && !shard.getNamespace().equals("public")) {
          throw new IllegalArgumentException(
              "Non-public namespaces are currently unsupported for PostgreSQL migrations.");
        }
      }
    }
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(SourceDbToSpannerOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    String workerMachineType =
        pipeline.getOptions().as(DataflowPipelineWorkerPoolOptions.class).getWorkerMachineType();
    Optional<Integer> resourceHintsMinCpus =
        DataflowWorkerMachineTypeUtils.getMinCpuResourceHint(pipeline.getOptions());
    if (!pipeline.getOptions().getRunner().getSimpleName().equals("DirectRunner")) {
      DataflowWorkerMachineTypeUtils.validateMachineSpecs(
          workerMachineType, 4, resourceHintsMinCpus);
    }

    SpannerConfig spannerConfig = createSpannerConfig(options);
    SourceConnectionConfig sourceConnectionConfig =
        PipelineController.getSourceConnectionConfig(
            options.getSourceDbDialect(), options.getSourceConfigURL());

    validateOptions(options, sourceConnectionConfig);

    // Decide type and source of migration
    ISrcToSpSourceConnector connector = SourceConnectorFactory.getSourceConnectorByDialect(options);
    return connector.executeMigration(options, sourceConnectionConfig, pipeline, spannerConfig);
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
            .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
            .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
            .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
    if (options.getMaxCommitDelay() >= 0) {
      spannerConfig = spannerConfig.withMaxCommitDelay(options.getMaxCommitDelay());
    }
    return spannerConfig;
  }
}