Plantilla de transmisión de datos a MySQL o PostgreSQL (Transmisión)

Beta

Esta función está sujeta a las “Condiciones de la Oferta de la Fase Previa a la DG” en la sección Condiciones del Servicio Generales de las Condiciones Específicas del Servicio. Las funciones de la fase previa a la DG están disponibles sin modificaciones y pueden tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

La plantilla de transmisión de datos a SQL es una canalización de transmisión que lee datos de Datastream y los replica en cualquier base de datos de MySQL o PostgreSQL. La plantilla lee los datos de Cloud Storage mediante notificaciones de Pub/Sub y los replica en tablas de réplica de PostgreSQL. Especifica el parámetro gcsPubSubSubscription para leer datos de las notificaciones de Pub/Sub O proporciona el parámetro inputFilePattern para leer datos directamente de los archivos en Cloud Storage.

La plantilla no es compatible con el lenguaje de definición de datos (DDL) y espera que todas las tablas ya existan en la base de datos. La replicación usa transformaciones con estado de Dataflow para filtrar los datos inactivos y garantizar la coherencia dentro de los datos desordenados. Por ejemplo, si ya se pasó una versión más reciente de una fila, se ignorará una versión tardía de esa fila. El lenguaje de manipulación de datos (DML) que se ejecuta es el mejor intento de replicar perfectamente los datos de origen o destino. Las declaraciones DML ejecutadas siguen las siguientes reglas:

• Si existe una clave primaria, las operaciones de inserción y actualización usan una sintaxis de upsert (es decir, INSERT INTO table VALUES (...) ON CONFLICT (...) DO UPDATE).
• Si las claves primarias existen, las eliminaciones se replican como un DML borrado.
• Si no existe una clave primaria, se insertan las operaciones de inserción y actualización en la tabla.
• Si no existen claves primarias, se ignoran las eliminaciones.

Si usas las utilidades de Oracle para Postgres, agrega ROWID en SQL como la clave primaria cuando no exista ninguna.

Requisitos de la canalización

• Una transmisión de Datastream que está lista para replicar los datos o ya los está replicando.
• Las notificaciones de Pub/Sub de Cloud Storage están habilitadas para los datos de Datastream.
• Se propagó una base de datos de PostgreSQL con el esquema requerido.
• Se configura el acceso a la red entre los trabajadores de Dataflow y PostgreSQL.

Parámetros de la plantilla

Parámetros obligatorios

• inputFilePattern: Es la ubicación del archivo para que los archivos de Datastream se repliquen en Cloud Storage. Esta ubicación suele ser la ruta de acceso raíz de la transmisión.
• databaseHost: Es el host de SQL al que se realizará la conexión.
• databaseUser: Es el usuario de SQL con todos los permisos necesarios para escribir en todas las tablas en la replicación.
• databasePassword: La contraseña para el usuario de SQL.

Parámetros opcionales

• gcsPubSubSubscription: Es la suscripción de Pub/Sub con las notificaciones de archivos de Datastream. Por ejemplo, projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_ID>
• inputFileFormat: El formato del archivo de salida que produce Datastream. Por ejemplo, avro o json. La configuración predeterminada es avro.
• streamName: El nombre o la plantilla del flujo que se consultará para obtener la información del esquema. El valor predeterminado es {_metadata_stream}.
• rfcStartDateTime: La fecha y hora de inicio que se usa para recuperar desde Cloud Storage (https://tools.ietf.org/html/rfc3339). El valor predeterminado es: 1970-01-01T00:00:00.00Z.
• dataStreamRootUrl: URL raíz de la API de Datastream. La configuración predeterminada es https://datastream.googleapis.com/.
• databaseType: El tipo de base de datos en la que se escribirá (por ejemplo, Postgres). La configuración predeterminada es: postgres.
• databasePort: Es el puerto de la base de datos de SQL al que se realizará la conexión. El valor predeterminado es 5432.
• databaseName: El nombre de la base de datos de SQL a la que se realizará la conexión. El valor predeterminado es postgres.
• defaultCasing: Es un botón de activación para el comportamiento de mayúsculas y minúsculas de la tabla. Por ejemplo,(es decir, LOWERCASE = mytable -> mytable, UPPERCASE = mytable -> MYTABLECAMEL = my_table -> myTable, SNAKE = myTable -> my_table. La configuración predeterminada es LOWERCASE.
• columnCasing: Es un botón de activación para el uso de mayúsculas y minúsculas en el nombre de la columna objetivo. LOWERCASE (predeterminado): my_column -> my_column. MAYÚSCULAS: my_column -> MY_COLUMN. CAMEL: my_column -> myColumn. SNAKE: myColumn -> my_column.
• schemaMap: Un mapa de claves o valores usados para dictar los cambios de nombre del esquema (es decir, old_name:new_name,CaseError:case_error). La configuración predeterminada es vacía.
• customConnectionString: Es la cadena de conexión opcional que se usará en lugar de la cadena de la base de datos predeterminada.
• numThreads: Determina el paralelismo clave del paso de formato a DML. Específicamente, el valor se pasa a Reshuffle.withNumBuckets. El valor predeterminado es 100.
• databaseLoginTimeout: Es el tiempo de espera en segundos para los intentos de acceso a la base de datos. Esto ayuda a evitar que se cuelgue la conexión cuando varios trabajadores intentan conectarse de forma simultánea.
• datastreamSourceType: Anula la detección del tipo de fuente para los datos de CDC de Datastream. Cuando se especifica, este valor se usará en lugar de derivar el tipo de fuente del campo read_method. Los valores válidos incluyen "mysql", "postgresql", "oracle", etcétera. Este parámetro es útil cuando el campo read_method contiene "cdc" y el tipo de fuente real no se puede determinar automáticamente.
• orderByIncludesIsDeleted: La ordenación por configuraciones de datos debe incluir la priorización de los datos que no se borraron. La configuración predeterminada es "false".

Ejecuta la plantilla

Console

1. Ve a la página Crear un trabajo a partir de una plantilla de Dataflow.
Ir a Crear un trabajo a partir de una plantilla
2. En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
3. Opcional: Para Extremo regional, selecciona un valor del menú desplegable. La región predeterminada es us-central1.

   Para obtener una lista de regiones en las que puedes ejecutar un trabajo de Dataflow, consulta Ubicaciones de Dataflow.

4. En el menú desplegable Plantilla de Dataflow, selecciona the Cloud Datastream to SQL template.
5. En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
6. Haga clic en Ejecutar trabajo.

gcloud

Nota: Para usar Google Cloud CLI a fin de ejecutar plantillas flex, debes tener Google Cloud CLI en la versión 284.0.0 o una versión posterior.

En tu shell o terminal, ejecuta la plantilla:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --enable-streaming-engine \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Cloud_Datastream_to_SQL \
    --parameters \
inputFilePattern=GCS_FILE_PATH,\
gcsPubSubSubscription=GCS_SUBSCRIPTION_NAME,\
databaseHost=DATABASE_HOST,\
databaseUser=DATABASE_USER,\
databasePassword=DATABASE_PASSWORD
  

Reemplaza lo siguiente:

• PROJECT_ID: El ID del proyecto Google Cloud en el que deseas ejecutar el trabajo de Dataflow
• JOB_NAME: Es el nombre del trabajo que elijas
• REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• VERSION: the version of the template that you want to use

  You can use the following values:

  • latest to use the latest version of the template, which is available in the non-dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/latest/
  • the version name, like 2023-09-12-00_RC00, to use a specific version of the template, which can be found nested in the respective dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/
  Caution: The latest version of templates might update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows.
• GCS_FILE_PATH: es la ruta de acceso de Cloud Storage a los datos de Datastream. Por ejemplo: gs://bucket/path/to/data/
• GCS_SUBSCRIPTION_NAME: es la suscripción de Pub/Sub desde la que se leen los archivos modificados. Por ejemplo: projects/my-project-id/subscriptions/my-subscription-id.
• DATABASE_HOST: Es la IP del host de SQL.
• DATABASE_USER: Es tu usuario de SQL.
• DATABASE_PASSWORD: Es tu contraseña de SQL.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud POST HTTP. Para obtener más información de la API y sus permisos de autorización, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {

          "inputFilePattern": "GCS_FILE_PATH",
          "gcsPubSubSubscription": "GCS_SUBSCRIPTION_NAME",
          "databaseHost": "DATABASE_HOST",
          "databaseUser": "DATABASE_USER",
          "databasePassword": "DATABASE_PASSWORD"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Cloud_Datastream_to_SQL",
   }
}
  

Reemplaza lo siguiente:

• PROJECT_ID: El ID del proyecto Google Cloud en el que deseas ejecutar el trabajo de Dataflow
• JOB_NAME: Es el nombre del trabajo que elijas
• LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• VERSION: the version of the template that you want to use

  You can use the following values:

  • latest to use the latest version of the template, which is available in the non-dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/latest/
  • the version name, like 2023-09-12-00_RC00, to use a specific version of the template, which can be found nested in the respective dated parent folder in the bucket— gs://dataflow-templates-REGION_NAME/
  Caution: The latest version of templates might update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows.
• GCS_FILE_PATH: es la ruta de acceso de Cloud Storage a los datos de Datastream. Por ejemplo: gs://bucket/path/to/data/
• GCS_SUBSCRIPTION_NAME: es la suscripción de Pub/Sub desde la que se leen los archivos modificados. Por ejemplo: projects/my-project-id/subscriptions/my-subscription-id.
• DATABASE_HOST: Es la IP del host de SQL.
• DATABASE_USER: Es tu usuario de SQL.
• DATABASE_PASSWORD: Es tu contraseña de SQL.

Código fuente de la plantilla

Java

/*
 * Copyright (C) 2020 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.metadata.TemplateParameter;
import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.datastream.io.CdcJdbcIO;
import com.google.cloud.teleport.v2.datastream.sources.DataStreamIO;
import com.google.cloud.teleport.v2.datastream.values.DmlInfo;
import com.google.cloud.teleport.v2.templates.DataStreamToSQL.Options;
import com.google.cloud.teleport.v2.transforms.CreateDml;
import com.google.cloud.teleport.v2.transforms.ProcessDml;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Splitter;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
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.options.StreamingOptions;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This pipeline ingests DataStream data from GCS. The data is then cleaned and converted from JSON
 * objects into DML statements. The DML is applied to the desired target database, which can be one
 * of MySQL or PostgreSQL. Replication maintains a 1:1 match between source and target by default.
 * No DDL is supported in the current version of this pipeline.
 *
 * <p>NOTE: Future versions will support: Pub/Sub, GCS, or Kafka as per DataStream
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/datastream-to-sql/README_Cloud_Datastream_to_SQL.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Cloud_Datastream_to_SQL",
    category = TemplateCategory.STREAMING,
    displayName = "Datastream to SQL",
    description = {
      "The Datastream to SQL template is a streaming pipeline that reads <a href=\"https://cloud.google.com/datastream/docs\">Datastream</a> data and replicates it into any MySQL or PostgreSQL database. "
          + "The template reads data from Cloud Storage using Pub/Sub notifications and replicates this data into SQL replica tables.\n",
      "The template does not support data definition language (DDL) and expects that all tables already exist in the database. "
          + "Replication uses Dataflow stateful transforms to filter stale data and ensure consistency in out of order data. "
          + "For example, if a more recent version of a row has already passed through, a late arriving version of that row is ignored. "
          + "The data manipulation language (DML) that executes is a best attempt to perfectly replicate source to target data. The DML statements executed follow the following rules:\n",
      "If a primary key exists, insert and update operations use upsert syntax (ie. <code>INSERT INTO table VALUES (...) ON CONFLICT (...) DO UPDATE</code>).\n"
          + "If primary keys exist, deletes are replicated as a delete DML.\n"
          + "If no primary key exists, both insert and update operations are inserted into the table.\n"
          + "If no primary keys exist, deletes are ignored.\n"
          + "If you are using the Oracle to Postgres utilities, add <code>ROWID</code> in SQL as the primary key when none exists."
    },
    optionsClass = Options.class,
    flexContainerName = "datastream-to-sql",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/datastream-to-sql",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "A Datastream stream that is ready to or already replicating data.",
      "<a href=\"https://cloud.google.com/storage/docs/reporting-changes\">Cloud Storage Pub/Sub notifications</a> are enabled for the Datastream data.",
      "A PostgreSQL database was seeded with the required schema.",
      "Network access between Dataflow workers and PostgreSQL is set up."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class DataStreamToSQL {

  private static final Logger LOG = LoggerFactory.getLogger(DataStreamToSQL.class);
  private static final String AVRO_SUFFIX = "avro";
  private static final String JSON_SUFFIX = "json";

  /**
   * Options supported by the pipeline.
   *
   * <p>Inherits standard configuration options.
   */
  public interface Options extends PipelineOptions, StreamingOptions {
    @TemplateParameter.GcsReadFile(
        order = 1,
        groupName = "Source",
        description = "File location for Datastream file input in Cloud Storage.",
        helpText =
            "The file location for the Datastream files in Cloud Storage to replicate. This file location is typically the root path for the stream.")
    String getInputFilePattern();

    void setInputFilePattern(String value);

    @TemplateParameter.PubsubSubscription(
        order = 2,
        optional = true,
        description = "The Pub/Sub subscription being used in a Cloud Storage notification policy.",
        helpText =
            "The Pub/Sub subscription with Datastream file notifications."
                + " For example, `projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_ID>`.")
    String getGcsPubSubSubscription();

    void setGcsPubSubSubscription(String value);

    @TemplateParameter.Enum(
        order = 3,
        enumOptions = {@TemplateEnumOption("avro"), @TemplateEnumOption("json")},
        optional = true,
        description = "Datastream output file format (avro/json).",
        helpText =
            "The format of the output file produced by Datastream. For example, `avro` or `json`. Defaults to `avro`.")
    @Default.String("avro")
    String getInputFileFormat();

    void setInputFileFormat(String value);

    @TemplateParameter.Text(
        order = 4,
        groupName = "Source",
        optional = true,
        description = "Name or template for the stream to poll for schema information.",
        helpText =
            "The name or template for the stream to poll for schema information. The default value is `{_metadata_stream}`.")
    String getStreamName();

    void setStreamName(String value);

    @TemplateParameter.DateTime(
        order = 5,
        optional = true,
        description =
            "The starting DateTime used to fetch from Cloud Storage "
                + "(https://tools.ietf.org/html/rfc3339).",
        helpText =
            "The starting DateTime used to fetch from Cloud Storage "
                + "(https://tools.ietf.org/html/rfc3339).")
    @Default.String("1970-01-01T00:00:00.00Z")
    String getRfcStartDateTime();

    void setRfcStartDateTime(String value);

    // DataStream API Root Url (only used for testing)
    @TemplateParameter.Text(
        order = 6,
        optional = true,
        description = "Datastream API Root URL (only required for testing)",
        helpText = "Datastream API Root URL")
    @Default.String("https://datastream.googleapis.com/")
    String getDataStreamRootUrl();

    void setDataStreamRootUrl(String value);

    // SQL Connection Parameters
    @TemplateParameter.Enum(
        order = 7,
        optional = true,
        enumOptions = {@TemplateEnumOption("postgres"), @TemplateEnumOption("mysql")},
        description = "SQL Database Type (postgres or mysql).",
        helpText = "The database type to write to (for example, Postgres).")
    @Default.String("postgres")
    String getDatabaseType();

    void setDatabaseType(String value);

    @TemplateParameter.Text(
        order = 8,
        groupName = "Target",
        description = "Database Host to connect on.",
        helpText = "The SQL host to connect on.")
    String getDatabaseHost();

    void setDatabaseHost(String value);

    @TemplateParameter.Text(
        order = 9,
        groupName = "Target",
        optional = true,
        description = "Database Port to connect on.",
        helpText = "The SQL database port to connect to. The default value is `5432`.")
    @Default.String("5432")
    String getDatabasePort();

    void setDatabasePort(String value);

    @TemplateParameter.Text(
        order = 10,
        description = "Database User to connect with.",
        helpText =
            "The SQL user with all required permissions to write to all tables in replication.")
    String getDatabaseUser();

    void setDatabaseUser(String value);

    @TemplateParameter.Password(
        order = 11,
        description = "Database Password for given user.",
        helpText = "The password for the SQL user.")
    String getDatabasePassword();

    void setDatabasePassword(String value);

    @TemplateParameter.Text(
        order = 12,
        groupName = "Target",
        optional = true,
        description = "SQL Database Name.",
        helpText = "The name of the SQL database to connect to. The default value is `postgres`.")
    @Default.String("postgres")
    String getDatabaseName();

    void setDatabaseName(String value);

    @TemplateParameter.Enum(
        order = 13,
        optional = true,
        enumOptions = {
          @TemplateEnumOption("LOWERCASE"),
          @TemplateEnumOption("UPPERCASE"),
          @TemplateEnumOption("CAMEL"),
          @TemplateEnumOption("SNAKE")
        },
        description = "Toggle for Table Casing",
        helpText =
            "A Toggle for table casing behavior. For example,(ie."
                + "LOWERCASE = mytable -> mytable, UPPERCASE = mytable -> MYTABLE"
                + "CAMEL = my_table -> myTable, SNAKE = myTable -> my_table")
    @Default.String("LOWERCASE")
    String getDefaultCasing();

    void setDefaultCasing(String value);

    @TemplateParameter.Enum(
        order = 14,
        optional = true,
        enumOptions = {
          @TemplateEnumOption("LOWERCASE"),
          @TemplateEnumOption("UPPERCASE"),
          @TemplateEnumOption("CAMEL"),
          @TemplateEnumOption("SNAKE")
        },
        description = "Toggle for Column Casing",
        helpText =
            "A toggle for target column name casing. "
                + "LOWERCASE (default): my_column -> my_column. "
                + "UPPERCASE: my_column -> MY_COLUMN. "
                + "CAMEL: my_column -> myColumn. "
                + "SNAKE: myColumn -> my_column.")
    @Default.String("LOWERCASE")
    String getColumnCasing();

    void setColumnCasing(String value);

    @TemplateParameter.Text(
        order = 15,
        optional = true,
        description = "A map of key/values used to dictate schema name changes",
        helpText =
            "A map of key/values used to dictate schema name changes (ie."
                + " old_name:new_name,CaseError:case_error)")
    @Default.String("")
    String getSchemaMap();

    void setSchemaMap(String value);

    @TemplateParameter.Text(
        order = 16,
        groupName = "Target",
        optional = true,
        description = "Custom connection string.",
        helpText =
            "Optional connection string which will be used instead of the default database string.")
    @Default.String("")
    String getCustomConnectionString();

    void setCustomConnectionString(String value);

    @TemplateParameter.Integer(
        order = 17,
        optional = true,
        description = "Number of threads to use for Format to DML step.",
        helpText =
            "Determines key parallelism of Format to DML step, specifically, the value is passed into Reshuffle.withNumBuckets.")
    @Default.Integer(100)
    int getNumThreads();

    void setNumThreads(int value);

    @TemplateParameter.Integer(
        order = 18,
        groupName = "Target",
        optional = true,
        description = "Database login timeout in seconds.",
        helpText =
            "The timeout in seconds for database login attempts. This helps prevent connection hangs when multiple workers try to connect simultaneously.")
    Integer getDatabaseLoginTimeout();

    void setDatabaseLoginTimeout(Integer value);

    @TemplateParameter.Boolean(
        order = 19,
        optional = true,
        description =
            "Order by configurations for data should include prioritizing data which is not deleted.",
        helpText =
            "Order by configurations for data should include prioritizing data which is not deleted.")
    @Default.Boolean(false)
    Boolean getOrderByIncludesIsDeleted();

    void setOrderByIncludesIsDeleted(Boolean value);

    @TemplateParameter.Text(
        order = 18,
        optional = true,
        description = "Datastream source type override",
        helpText =
            "Override the source type detection for Datastream CDC data. When specified, this value will be used instead of deriving the source type from the read_method field. Valid values include 'mysql', 'postgresql', 'oracle', etc. This parameter is useful when the read_method field contains 'cdc' and the actual source type cannot be determined automatically.")
    String getDatastreamSourceType();

    void setDatastreamSourceType(String value);
  }

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

    LOG.info("Starting Datastream to SQL");

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

    options.setStreaming(true);
    run(options);
  }

  /**
   * Build the DataSourceConfiguration for the target SQL database. Using the pipeline options,
   * determine the database type and create the correct jdbc connection for the requested DB.
   *
   * @param options The execution parameters to the pipeline.
   */
  public static CdcJdbcIO.DataSourceConfiguration getDataSourceConfiguration(Options options) {
    String jdbcDriverName;
    String jdbcDriverConnectionString;

    switch (options.getDatabaseType()) {
      case "postgres":
        jdbcDriverName = "org.postgresql.Driver";
        jdbcDriverConnectionString =
            String.format(
                "jdbc:postgresql://%s:%s/%s",
                options.getDatabaseHost(), options.getDatabasePort(), options.getDatabaseName());
        break;
      case "mysql":
        jdbcDriverName = "com.mysql.cj.jdbc.Driver";
        jdbcDriverConnectionString =
            String.format(
                "jdbc:mysql://%s:%s/%s",
                options.getDatabaseHost(), options.getDatabasePort(), options.getDatabaseName());
        break;
      default:
        throw new IllegalArgumentException(
            String.format("Database Type %s is not supported.", options.getDatabaseType()));
    }
    if (!options.getCustomConnectionString().isEmpty()) {
      jdbcDriverConnectionString = options.getCustomConnectionString();
    }

    CdcJdbcIO.DataSourceConfiguration dataSourceConfiguration =
        CdcJdbcIO.DataSourceConfiguration.create(jdbcDriverName, jdbcDriverConnectionString)
            .withUsername(options.getDatabaseUser())
            .withPassword(options.getDatabasePassword())
            .withMaxIdleConnections(new Integer(0))
            .withLoginTimeout(options.getDatabaseLoginTimeout());

    return dataSourceConfiguration;
  }

  /**
   * Validate the options supplied match expected values. We will also validate that connectivity is
   * working correctly for the target SQL database.
   *
   * @param options The execution parameters to the pipeline.
   * @param dataSourceConfiguration The JDBC datasource configuration.
   */
  public static void validateOptions(
      Options options, CdcJdbcIO.DataSourceConfiguration dataSourceConfiguration) {
    try {
      if (options.getDatabaseHost() != null) {
        dataSourceConfiguration.buildDatasource().getConnection().close();
      }
    } catch (SQLException e) {
      throw new IllegalArgumentException(e);
    }
  }

  /**
   * Parses a single map string and resolves it into schema and table mappings, intelligently
   * inferring a schema-to-schema mapping if only table-specific rules are provided.
   *
   * @param mappingString A comma-separated string of mapping rules.
   */
  public static Map<String, Map<String, String>> parseMappings(String mappingString) {
    Map<String, String> schemaMappings = new HashMap<>();
    Map<String, String> tableMappings = new HashMap<>();

    if (mappingString != null && !mappingString.isEmpty()) {
      Map<String, String> allMappings =
          Splitter.on("|").withKeyValueSeparator(":").split(mappingString);

      // Strictly separate rules based on the presence of a dot.
      for (Map.Entry<String, String> entry : allMappings.entrySet()) {
        if (entry.getKey().contains(".")) {
          tableMappings.put(entry.getKey(), entry.getValue());
        } else {
          schemaMappings.put(entry.getKey(), entry.getValue());
        }
      }
    }
    Map<String, Map<String, String>> mappings = new HashMap<>();
    mappings.put("schemas", schemaMappings);
    mappings.put("tables", tableMappings);
    return mappings;
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  public static PipelineResult run(Options options) {
    /*
     * Stages:
     * 1) Ingest and Normalize Data to FailsafeElement with JSON Strings
     * 2) Write JSON Strings to SQL DML Objects
     * 3) Filter stale rows using stateful PK transform
     * 4) Write DML statements to SQL Database via jdbc
     */

    Pipeline pipeline = Pipeline.create(options);

    CdcJdbcIO.DataSourceConfiguration dataSourceConfiguration = getDataSourceConfiguration(options);
    validateOptions(options, dataSourceConfiguration);

    Map<String, Map<String, String>> mappings = parseMappings(options.getSchemaMap());
    Map<String, String> schemaMap = mappings.get("schemas");
    Map<String, String> tableNameMap = mappings.get("tables");

    LOG.info("Parsed schema map: {}", schemaMap);
    LOG.info("Parsed table name map: {}", tableNameMap);

    /*
     * Stage 1: Ingest and Normalize Data to FailsafeElement with JSON Strings
     * a) Read DataStream data from GCS into JSON String FailsafeElements (datastreamJsonRecords)
     */
    PCollection<FailsafeElement<String, String>> datastreamJsonRecords =
        pipeline.apply(
            new DataStreamIO(
                    options.getStreamName(),
                    options.getInputFilePattern(),
                    options.getInputFileFormat(),
                    options.getGcsPubSubSubscription(),
                    options.getRfcStartDateTime())
                .withRenameColumnValue("_metadata_row_id", "rowid")
                .withHashRowId()
                .withDatastreamSourceType(options.getDatastreamSourceType()));

    /*
     * Stage 2: Write JSON Strings to SQL Insert Strings
     * a) Convert JSON String FailsafeElements to TableRow's (tableRowRecords)
     * Stage 3) Filter stale rows using stateful PK transform
     */
    PCollection<KV<String, DmlInfo>> dmlStatements =
        datastreamJsonRecords
            .apply(
                "Format to DML",
                CreateDml.of(dataSourceConfiguration)
                    .withDefaultCasing(options.getDefaultCasing())
                    .withSchemaMap(schemaMap)
                    .withTableNameMap(tableNameMap)
                    .withColumnCasing(options.getColumnCasing())
                    .withOrderByIncludesIsDeleted(options.getOrderByIncludesIsDeleted())
                    .withNumThreads(options.getNumThreads()))
            .apply("DML Stateful Processing", ProcessDml.statefulOrderByPK());

    /*
     * Stage 4: Write Inserts to CloudSQL
     */
    dmlStatements.apply(
        "Write to SQL",
        CdcJdbcIO.<KV<String, DmlInfo>>write()
            .withDataSourceConfiguration(dataSourceConfiguration)
            .withStatementFormatter(
                new CdcJdbcIO.StatementFormatter<KV<String, DmlInfo>>() {
                  public String formatStatement(KV<String, DmlInfo> element) {
                    LOG.debug("Executing SQL: {}", element.getValue().getDmlSql());
                    return element.getValue().getDmlSql();
                  }
                }));

    // Execute the pipeline and return the result.
    return pipeline.run();
  }
}

¿Qué sigue?

• Obtén información sobre las plantillas de Dataflow.
• Consulta la lista de plantillas que proporciona Google.