Vorlage „Sourcedb zu Spanner“

Die Vorlage "SourceDB zu Spanner" ist eine Batchpipeline, die Daten aus einer relationalen Datenbank in eine vorhandene Spanner-Datenbank kopiert. Diese Pipeline verwendet JDBC, um eine Verbindung zur relationalen Datenbank herzustellen. Sie können mit dieser Vorlage Daten aus einer beliebigen relationalen Datenbank mit den verfügbaren JDBC-Treibern nach Spanner kopieren. Es werden nur eine begrenzte Anzahl von MySQL-Typen unterstützt.

Als zusätzliche Schutzmaßnahme können Sie einen Cloud KMS-Schlüssel zusammen mit einem Base64-codierten Nutzernamen, Passwort und Verbindungsstring-Parametern übergeben, die mit dem Cloud KMS-Schlüssel verschlüsselt sind. Weitere Informationen zum Verschlüsseln von Nutzernamen, Passwörtern und Verbindungsstring-Parametern finden Sie unter Cloud KMS API-Verschlüsselungsendpunkt.

Pipelineanforderungen

Die JDBC-Treiber für die relationale Datenbank müssen verfügbar sein.
Die Spanner-Tabellen müssen vor der Pipelineausführung vorhanden sein.
Die Spanner-Tabellen müssen ein kompatibles Schema haben.
Die relationale Datenbank muss über das Subnetz zugänglich sein, in dem Dataflow ausgeführt wird.

Vorlagenparameter

Parameter	Beschreibung
`sourceConfigURL`	Der URL-String für die JDBC-Verbindung. Beispiel: `jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8` oder die Shard-Konfiguration.
`instanceId`	Die Cloud Spanner-Zielinstanz.
`databaseId`	Die Ziel-Cloud Spanner-Datenbank.
`projectId`	Dies ist der Name des Cloud Spanner-Projekts.
`outputDirectory`	Dieses Verzeichnis wird verwendet, um die fehlgeschlagenen/übersprungenen/gefilterten Datensätze bei einer Migration zu verschieben.
`jdbcDriverJars`	Optional: Die durch kommagetrennte Liste der JAR-Dateien des Treibers. Beispiel: `gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar`. Die Standardeinstellung ist leer.
`jdbcDriverClassName`	Optional: Der Name der JDBC-Treiberklasse. Beispiel: `com.mysql.jdbc.Driver`. Standardeinstellung: `com`.mysql.jdbc.Driver.
`username`	Optional: Der Nutzername, der für die JDBC-Verbindung verwendet werden soll. Die Standardeinstellung ist leer.
`password`	Optional: Das Passwort für die JDBC-Verbindung. Die Standardeinstellung ist leer.
`tables`	Optional: Tabellen, die aus der Quelle migriert werden sollen. Die Standardeinstellung ist leer.
`numPartitions`	Optional: Die Anzahl der Partitionen. Damit bilden zusammen mit der Unter- und Obergrenze die Partitionsschritte für generierte WHERE-Klauselausdrücke, die zum gleichmäßigen Aufteilen der Partitionsspalte verwendet werden. Wenn die Eingabe kleiner als 1 ist, wird die Zahl auf 1 gesetzt. Die Standardeinstellung ist `0`.
`spannerHost`	Optional: Der Cloud Spanner-Endpunkt, der in der Vorlage aufgerufen werden soll. Beispiel: `https://batch-spanner.googleapis.com`. Standardeinstellung: `https://batch-spanner`.googleapis.com.
`maxConnections`	Optional: Konfiguriert den JDBC-Verbindungspool auf jedem Worker mit der maximalen Anzahl von Verbindungen. Verwenden Sie eine negative Zahl, wenn Sie kein Limit haben. Beispiel: `-1`. Die Standardeinstellung ist `0`.
`sessionFilePath`	Optional: Sitzungspfad in Cloud Storage, der Zuordnungsinformationen aus dem Spanner-Migrationstool enthält. Die Standardeinstellung ist leer.
`transformationJarPath`	Optional: Benutzerdefinierter JAR-Speicherort in Cloud Storage, der die benutzerdefinierte Transformationslogik für die Verarbeitung von Einträgen enthält. Die Standardeinstellung ist leer.
`transformationClassName`	Optional: Vollständig qualifizierter Klassenname mit benutzerdefinierter Transformationslogik. Dies ist ein Pflichtfeld, falls transformationJarPath angegeben ist. Die Standardeinstellung ist leer.
`transformationCustomParameters`	Optional: String mit allen benutzerdefinierten Parametern, die an die benutzerdefinierte Transformationsklasse übergeben werden sollen. Die Standardeinstellung ist leer.
`disabledAlgorithms`	Optional: Durch Kommas getrennte Algorithmen zum Deaktivieren. Wenn dieser Wert auf „Keine“ gesetzt ist, wird kein Algorithmus deaktiviert. Verwenden Sie diesen Parameter mit Vorsicht, da die standardmäßig deaktivierten Algorithmen Sicherheitslücken oder Leistungsprobleme haben können. Beispiel: `SSLv3, RC4`.
`extraFilesToStage`	Optional: Durch Kommas getrennte Cloud Storage-Pfade oder Secret Manager-Secrets für Dateien, die im Worker bereitgestellt werden sollen. Diese Dateien werden im Verzeichnis /extra_files in jedem Worker gespeichert. Beispiel: `gs://<BUCKET>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>`.

Führen Sie die Vorlage aus.

Console

Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.

Zur Seite "Job aus Vorlage erstellen“

Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.
Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.
Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the Sourcedb to Spanner templateaus.
Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
Klicken Sie auf Job ausführen.

gcloud-CLI

Führen Sie die Vorlage in der Shell oder im Terminal aus:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/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,\

Ersetzen Sie dabei Folgendes:

JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
VERSION: Die Version der Vorlage, die Sie verwenden möchten
Sie können die folgenden Werte verwenden:
- latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
- Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
SOURCE_CONFIG_URL: URL zum Herstellen einer Verbindung zum Quelldatenbankhost. Es kann Folgendes sein: 1. Die JDBC-Verbindungs-URL, die den Host, den Port und den Namen der Quelldatenbank enthalten muss und optional Eigenschaften wie „autoReconnect“ und „maxReconnects“ enthalten kann. Format: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Der Pfad zur Shard-Konfiguration
INSTANCE_ID: Die Cloud Spanner-Instanz-ID.
DATABASE_ID: die Cloud Spanner-Datenbank-ID
PROJECT_ID: die Cloud Spanner-Projekt-ID.
OUTPUT_DIRECTORY: das Ausgabeverzeichnis für fehlgeschlagene/übersprungene/gefilterte Ereignisse

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter 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-LOCATION/VERSION/flex/Sourcedb_to_Spanner_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

Ersetzen Sie dabei Folgendes:

PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
VERSION: Die Version der Vorlage, die Sie verwenden möchten
Sie können die folgenden Werte verwenden:
- latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
- Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
SOURCE_CONFIG_URL: URL zum Herstellen einer Verbindung zum Quelldatenbankhost. Es kann Folgendes sein: 1. Die JDBC-Verbindungs-URL, die den Host, den Port und den Namen der Quelldatenbank enthalten muss und optional Eigenschaften wie „autoReconnect“ und „maxReconnects“ enthalten kann. Format: `jdbc:mysql://{host}:{port}/{dbName}?{parameters}`2. Der Pfad zur Shard-Konfiguration
INSTANCE_ID: Die Cloud Spanner-Instanz-ID.
DATABASE_ID: die Cloud Spanner-Datenbank-ID
PROJECT_ID: die Cloud Spanner-Projekt-ID.
OUTPUT_DIRECTORY: das Ausgabeverzeichnis für fehlgeschlagene/übersprungene/gefilterte Ereignisse

Quellcode der Vorlage

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.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl;
import com.google.cloud.teleport.v2.spanner.migrations.utils.ShardFileReader;
import com.google.common.annotations.VisibleForTesting;
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.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 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",
      "For an extra layer of protection, you can also pass in a Cloud KMS key along with a"
          + " Base64-encoded username, password, and connection string parameters encrypted with"
          + " the Cloud KMS key. See the <a"
          + " href=\"https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt\">Cloud"
          + " KMS API encryption endpoint</a> for additional details on encrypting your username,"
          + " password, and connection string parameters."
    },
    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 {

  private static final Logger LOG = LoggerFactory.getLogger(SourceDbToSpanner.class);

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

    // Parse the user options passed from the command-line
    SourceDbToSpannerOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SourceDbToSpannerOptions.class);
    run(options);
  }

  /**
   * 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) {
    // TODO - Validate if options are as expected
    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig = createSpannerConfig(options);

    // Decide type and source of migration
    if (options.getSourceConfigURL().startsWith("gs://")) {
      List<Shard> shards =
          new ShardFileReader(new SecretManagerAccessorImpl())
              .readForwardMigrationShardingConfig(options.getSourceConfigURL());
      return PipelineController.executeShardedMigration(options, pipeline, shards, spannerConfig);
    } else {
      return PipelineController.executeSingleInstanceMigration(options, pipeline, spannerConfig);
    }
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
    return 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()));
  }
}