Plantillas que proporciona Google | Cloud Dataflow

Google ofrece un conjunto de plantillas de código abierto de Cloud Dataflow. Estas son las plantillas disponibles:

WordCount
Cloud Bigtable a Cloud Storage AvroFile
Cloud Bigtable a Cloud Storage SequenceFile
Cloud Pub/Sub a BigQuery
Cloud Storage Text a Cloud Pub/Sub (lote)
Cloud Storage Text a Cloud Pub/Sub (transmisión)
Cloud Pub/Sub a Cloud Storage Text
Cloud Pub/Sub a Cloud Pub/Sub
Cloud Datastore a Cloud Storage Text
Cloud Storage Text a BigQuery
Cloud Storage Text a Cloud Datastore
Cloud Storage AvroFile a Cloud Bigtable
Cloud Storage SequenceFile a Cloud Bigtable
Compresión masiva de archivos de Cloud Storage
Descompresión masiva de archivos de Cloud Storage
Borrado masivo de Cloud Datastore
Cloud Spanner a Cloud Storage Avro
Cloud Spanner a Cloud Storage Text
Cloud Storage Avro a Cloud Spanner
Java Database Connectivity (JDBC) a BigQuery
Cloud Pub/Sub a Cloud Storage Avro

WordCount

La plantilla de WordCount es una canalización por lotes que lee texto de Cloud Storage, convierte las líneas de texto en tokens de palabras individuales y cuenta la frecuencia con la que aparece cada palabra. Para obtener más información sobre WordCount, consulta Canalización de ejemplo de WordCount.

Parámetros de la plantilla

Parámetro	Descripción
`inputFile`	Ruta de acceso del archivo de entrada de Cloud Storage.
`output`	Ruta de acceso y prefijo del archivo de salida de Cloud Storage.

Ejecuta la plantilla de WordCount

Console

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe WordCount templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/latest/Word_Count

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Word_Count \
    --parameters \
inputFile=gs://dataflow-samples/shakespeare/kinglear.txt,\
output=gs://YOUR_BUCKET_NAME/output/my_output

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/latest/Word_Count

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Word_Count
{
    "jobName": "JOB_NAME",
    "parameters": {
       "inputFile" : "gs://dataflow-samples/shakespeare/kinglear.txt",
       "output": "gs://YOUR_BUCKET_NAME/output/my_output"
    },
    "environment": { "zone": "us-central1-f" }
}

Cloud Bigtable a Cloud Storage AvroFile

La plantilla de Cloud Bigtable a Cloud Storage AvroFile es una canalización que lee datos de la tabla de Cloud Bigtable y los escribe en el depósito de Cloud Storage en formato AvroFile. Puedes usar la plantilla como una solución rápida para mover datos desde Cloud Bigtable hacia Cloud Storage.

Requisitos para esta canalización:

La tabla de Cloud Bigtable ya debe existir.
El depósito de salida de Cloud Storage ya debe existir antes de ejecutar la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`bigtableProjectId`	ID del proyecto de GCP en la instancia de Cloud Bigtable desde la que quieres leer datos.
`bigtableInstanceId`	ID de la instancia de Cloud Bigtable que contiene la tabla.
`bigtableTableId`	ID de la tabla de Cloud Bigtable para exportar.
`outputDirectory`	Ruta de acceso de Cloud Storage donde se escriben los datos. Por ejemplo, `gs://mybucket/somefolder`.
`filenamePrefix`	Prefijo del nombre del archivo AvroFile. Por ejemplo, `output-`.

Ejecuta la plantilla de Cloud Bigtable a Cloud Storage AvroFile

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Selecciona the Cloud Bigtable to Cloud Storage AvroFile templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_AvroFile

Utiliza esta solicitud de ejemplo como se describe en Usa la API de REST. Esta solicitud requiere autorización, y debes especificar una tempLocation en la que tengas permisos de escritura. En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [OUTPUT_DIRECTORY] con la ruta de acceso de Cloud Storage donde se deben escribir los datos. Por ejemplo, gs://mybucket/somefolder.
Reemplaza [FILENAME_PREFIX] con el prefijo del nombre del archivo AvroFile. Por ejemplo, output-.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_AvroFile \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],outputDirectory=[OUTPUT_DIRECTORY],filenamePrefix=[FILENAME_PREFIX]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_AvroFile

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [OUTPUT_DIRECTORY] con la ruta de acceso de Cloud Storage donde se deben escribir los datos. Por ejemplo, gs://mybucket/somefolder.
Reemplaza [FILENAME_PREFIX] con el prefijo del nombre del archivo AvroFile. Por ejemplo, output-.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_AvroFile
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "outputDirectory": "[OUTPUT_DIRECTORY]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/DataflowTemplates de GitHub.

Python

El código fuente de Python no está disponible.

Cloud Bigtable a Cloud Storage SequenceFile

La plantilla de Cloud Bigtable a Cloud Storage SequenceFile es una canalización que lee datos de la tabla de Cloud Bigtable y los escribe en el depósito de Cloud Storage en formato SequenceFile. Puedes usar la plantilla como una solución rápida para mover datos desde Cloud Bigtable hacia Cloud Storage.

Requisitos para esta canalización:

La tabla de Cloud Bigtable ya debe existir.
El depósito de salida de Cloud Storage ya debe existir antes de ejecutar la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`bigtableProject`	ID del proyecto de GCP en la instancia de Cloud Bigtable desde la que quieres leer datos.
`bigtableInstanceId`	ID de la instancia de Cloud Bigtable que contiene la tabla.
`bigtableTableId`	ID de la tabla de Cloud Bigtable para exportar.
`bigtableAppProfileId`	ID del perfil de la aplicación de Cloud Bigtable que se usa para exportar. Si no especificas un perfil de aplicación, Cloud Bigtable utiliza el perfil predeterminado de la aplicación de la instancia.
`destinationPath`	Ruta de acceso de Cloud Storage donde se escriben los datos. Por ejemplo, `gs://mybucket/somefolder`.
`filenamePrefix`	Prefijo del nombre del archivo SequenceFile. Por ejemplo, `output-`.

Ejecuta la plantilla de Cloud Bigtable a Cloud Storage SequenceFile

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Bigtable to Cloud Storage SequenceFile templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_SequenceFile

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [APPLICATION_PROFILE_ID] con el ID del perfil de la aplicación de Cloud Bigtable que se usa para exportar.
Reemplaza [DESTINATION_PATH] con la ruta de acceso de Cloud Storage donde se deben escribir los datos. Por ejemplo, gs://mybucket/somefolder.
Reemplaza [FILENAME_PREFIX] con el prefijo del nombre del archivo SequenceFile. Por ejemplo, output-.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_SequenceFile \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],destinationPath=[DESTINATION_PATH],filenamePrefix=[FILENAME_PREFIX]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_SequenceFile

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [APPLICATION_PROFILE_ID] con el ID del perfil de la aplicación de Cloud Bigtable que se usa para exportar.
Reemplaza [DESTINATION_PATH] con la ruta de acceso de Cloud Storage donde se deben escribir los datos. Por ejemplo, gs://mybucket/somefolder.
Reemplaza [FILENAME_PREFIX] con el prefijo del nombre del archivo SequenceFile. Por ejemplo, output-.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_SequenceFile
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "destinationPath": "[DESTINATION_PATH]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/cloud-bigtable-client de GitHub.

Python

El código fuente de Python no está disponible.

Cloud Pub/Sub a BigQuery

La plantilla de Cloud Pub/Sub a BigQuery es una canalización de transmisión que lee mensajes con formato JSON desde un tema de Cloud Pub/Sub y los escribe en una tabla de BigQuery. Puedes usar la plantilla como una solución rápida para mover datos desde Cloud Pub/Sub hacia BigQuery. La plantilla lee mensajes con formato JSON desde Cloud Pub/Sub y los convierte en elementos de BigQuery.

Requisitos para esta canalización:

Los mensajes de Cloud Pub/Sub deben tener formato JSON, como se describe aquí. Por ejemplo, los mensajes con formato {"k1":"v1", "k2":"v2"} se pueden insertar en una tabla de BigQuery con dos columnas, llamadas k1 y k2, con tipo de datos de string.
La tabla de salida debe existir antes de que se ejecute la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`inputTopic`	Tema de entrada de Cloud Pub/Sub para leer, con el formato `projects/<project>/topics/<topic>`.
`outputTableSpec`	Ubicación de la tabla de salida de BigQuery, con el formato `<my-project>:<my-dataset>.<my-table>`

Ejecuta la plantilla de Cloud Pub/Sub a BigQuery

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Selecciona the Cloud Pub/Sub to BigQuery template en el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/PubSub_to_BigQuery

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza YOUR_DATASET con tu conjunto de datos de BigQuery y YOUR_TABLE_NAME con el nombre de tu tabla de BigQuery.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/PubSub_to_BigQuery \
    --parameters \
inputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME,\
outputTableSpec=YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/PubSub_to_BigQuery

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza YOUR_DATASET con tu conjunto de datos de BigQuery y YOUR_TABLE_NAME con el nombre de tu tabla de BigQuery.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/PubSub_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME",
       "outputTableSpec": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.coders.FailsafeElementCoder;
import com.google.cloud.teleport.templates.common.BigQueryConverters.FailsafeJsonToTableRow;
import com.google.cloud.teleport.templates.common.ErrorConverters.WritePubsubMessageErrors;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.FailsafeJavascriptUdf;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.util.DualInputNestedValueProvider;
import com.google.cloud.teleport.util.DualInputNestedValueProvider.TranslatorInput;
import com.google.cloud.teleport.util.ResourceUtils;
import com.google.cloud.teleport.util.ValueProviderUtils;
import com.google.cloud.teleport.values.FailsafeElement;
import java.nio.charset.StandardCharsets;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessageWithAttributesCoder;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link PubSubToBigQuery} pipeline is a streaming pipeline which ingests data in JSON format
 * from Cloud Pub/Sub, executes a UDF, and outputs the resulting records to BigQuery. Any errors
 * which occur in the transformation of the data or execution of the UDF will be output to a
 * separate errors table in BigQuery. The errors table will be created if it does not exist prior to
 * execution. Both output and error tables are specified by the user as template parameters.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>The Pub/Sub topic exists.
 *   <li>The BigQuery output table exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT_ID=PROJECT ID HERE
 * BUCKET_NAME=BUCKET NAME HERE
 * PIPELINE_FOLDER=gs://${BUCKET_NAME}/dataflow/pipelines/pubsub-to-bigquery
 *
 * # Set the runner
 * RUNNER=DataflowRunner
 *
 * # Build the template
 * mvn compile exec:java \
 * -Dexec.mainClass=com.google.cloud.teleport.templates.PubSubToBigQuery \
 * -Dexec.cleanupDaemonThreads=false \
 * -Dexec.args=" \
 * --project=${PROJECT_ID} \
 * --stagingLocation=${PIPELINE_FOLDER}/staging \
 * --tempLocation=${PIPELINE_FOLDER}/temp \
 * --templateLocation=${PIPELINE_FOLDER}/template \
 * --runner=${RUNNER}"
 *
 * # Execute the template
 * JOB_NAME=pubsub-to-bigquery-$USER-`date +"%Y%m%d-%H%M%S%z"`
 *
 * gcloud dataflow jobs run ${JOB_NAME} \
 * --gcs-location=${PIPELINE_FOLDER}/template \
 * --zone=us-east1-d \
 * --parameters \
 * "inputTopic=projects/data-analytics-pocs/topics/teleport-pubsub-to-bigquery,\
 * outputTableSpec=data-analytics-pocs:demo.pubsub_to_bigquery,\
 * outputDeadletterTable=data-analytics-pocs:demo.pubsub_to_bigquery_deadletter"
 * </pre>
 */
public class PubSubToBigQuery {

  /** The log to output status messages to. */
  private static final Logger LOG = LoggerFactory.getLogger(PubSubToBigQuery.class);

  /** The tag for the main output for the UDF. */
  public static final TupleTag<FailsafeElement<PubsubMessage, String>> UDF_OUT =
      new TupleTag<FailsafeElement<PubsubMessage, String>>() {};

  /** The tag for the main output of the json transformation. */
  public static final TupleTag<TableRow> TRANSFORM_OUT = new TupleTag<TableRow>() {};

  /** The tag for the dead-letter output of the udf. */
  public static final TupleTag<FailsafeElement<PubsubMessage, String>> UDF_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<PubsubMessage, String>>() {};

  /** The tag for the dead-letter output of the json to table row transform. */
  public static final TupleTag<FailsafeElement<PubsubMessage, String>> TRANSFORM_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<PubsubMessage, String>>() {};

  /** The default suffix for error tables if dead letter table is not specified. */
  public static final String DEFAULT_DEADLETTER_TABLE_SUFFIX = "_error_records";

  /**
   * The {@link Options} class provides the custom execution options passed by the executor at the
   * command-line.
   */
  public interface Options extends PipelineOptions, JavascriptTextTransformerOptions {
    @Description("Table spec to write the output to")
    ValueProvider<String> getOutputTableSpec();

    void setOutputTableSpec(ValueProvider<String> value);

    @Description("Pub/Sub topic to read the input from")
    ValueProvider<String> getInputTopic();

    void setInputTopic(ValueProvider<String> value);

    @Description(
        "The dead-letter table to output to within BigQuery in <project-id>:<dataset>.<table> "
            + "format. If it doesn't exist, it will be created during pipeline execution.")
    ValueProvider<String> getOutputDeadletterTable();

    void setOutputDeadletterTable(ValueProvider<String> value);
  }

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * PubSubToBigQuery#run(Options)} method to start the pipeline and invoke {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {

    Pipeline pipeline = Pipeline.create(options);

    // Register the coder for pipeline
    FailsafeElementCoder<PubsubMessage, String> coder =
        FailsafeElementCoder.of(PubsubMessageWithAttributesCoder.of(), StringUtf8Coder.of());

    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(coder.getEncodedTypeDescriptor(), coder);

    /*
     * Steps:
     *  1) Read messages in from Pub/Sub
     *  2) Transform the PubsubMessages into TableRows
     *     - Transform message payload via UDF
     *     - Convert UDF result to TableRow objects
     *  3) Write successful records out to BigQuery
     *  4) Write failed records out to BigQuery
     */
    PCollectionTuple transformOut =
        pipeline
            /*
             * Step #1: Read messages in from Pub/Sub
             */
            .apply(
                "ReadPubsubMessages",
                PubsubIO.readMessagesWithAttributes().fromTopic(options.getInputTopic()))

            /*
             * Step #2: Transform the PubsubMessages into TableRows
             */
            .apply("ConvertMessageToTableRow", new PubsubMessageToTableRow(options));

    /*
     * Step #3: Write the successful records out to BigQuery
     */
    transformOut
        .get(TRANSFORM_OUT)
        .apply(
            "WriteSuccessfulRecords",
            BigQueryIO.writeTableRows()
                .withoutValidation()
                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                .to(options.getOutputTableSpec()));

    /*
     * Step #4: Write failed records out to BigQuery
     */
    PCollectionList.of(transformOut.get(UDF_DEADLETTER_OUT))
        .and(transformOut.get(TRANSFORM_DEADLETTER_OUT))
        .apply("Flatten", Flatten.pCollections())
        .apply(
            "WriteFailedRecords",
            WritePubsubMessageErrors.newBuilder()
                .setErrorRecordsTable(
                    ValueProviderUtils.maybeUseDefaultDeadletterTable(
                        options.getOutputDeadletterTable(),
                        options.getOutputTableSpec(),
                        DEFAULT_DEADLETTER_TABLE_SUFFIX))
                .setErrorRecordsTableSchema(ResourceUtils.getDeadletterTableSchemaJson())
                .build());

    return pipeline.run();
  }

  /**
   * If deadletterTable is available, it is returned as is, otherwise outputTableSpec +
   * defaultDeadLetterTableSuffix is returned instead.
   */
  private static ValueProvider<String> maybeUseDefaultDeadletterTable(
      ValueProvider<String> deadletterTable,
      ValueProvider<String> outputTableSpec,
      String defaultDeadLetterTableSuffix) {
    return DualInputNestedValueProvider.of(
        deadletterTable,
        outputTableSpec,
        new SerializableFunction<TranslatorInput<String, String>, String>() {
          @Override
          public String apply(TranslatorInput<String, String> input) {
            String userProvidedTable = input.getX();
            String outputTableSpec = input.getY();
            if (userProvidedTable == null) {
              return outputTableSpec + defaultDeadLetterTableSuffix;
            }
            return userProvidedTable;
          }
        });
  }

  /**
   * The {@link PubsubMessageToTableRow} class is a {@link PTransform} which transforms incoming
   * {@link PubsubMessage} objects into {@link TableRow} objects for insertion into BigQuery while
   * applying an optional UDF to the input. The executions of the UDF and transformation to {@link
   * TableRow} objects is done in a fail-safe way by wrapping the element with it's original payload
   * inside the {@link FailsafeElement} class. The {@link PubsubMessageToTableRow} transform will
   * output a {@link PCollectionTuple} which contains all output and dead-letter {@link
   * PCollection}.
   *
   * <p>The {@link PCollectionTuple} output will contain the following {@link PCollection}:
   *
   * <ul>
   *   <li>{@link PubSubToBigQuery#UDF_OUT} - Contains all {@link FailsafeElement} records
   *       successfully processed by the optional UDF.
   *   <li>{@link PubSubToBigQuery#UDF_DEADLETTER_OUT} - Contains all {@link FailsafeElement}
   *       records which failed processing during the UDF execution.
   *   <li>{@link PubSubToBigQuery#TRANSFORM_OUT} - Contains all records successfully converted from
   *       JSON to {@link TableRow} objects.
   *   <li>{@link PubSubToBigQuery#TRANSFORM_DEADLETTER_OUT} - Contains all {@link FailsafeElement}
   *       records which couldn't be converted to table rows.
   * </ul>
   */
  static class PubsubMessageToTableRow
      extends PTransform<PCollection<PubsubMessage>, PCollectionTuple> {

    private final Options options;

    PubsubMessageToTableRow(Options options) {
      this.options = options;
    }

    @Override
    public PCollectionTuple expand(PCollection<PubsubMessage> input) {

      PCollectionTuple udfOut =
          input
              // Map the incoming messages into FailsafeElements so we can recover from failures
              // across multiple transforms.
              .apply("MapToRecord", ParDo.of(new PubsubMessageToFailsafeElementFn()))
              .apply(
                  "InvokeUDF",
                  FailsafeJavascriptUdf.<PubsubMessage>newBuilder()
                      .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                      .setFunctionName(options.getJavascriptTextTransformFunctionName())
                      .setSuccessTag(UDF_OUT)
                      .setFailureTag(UDF_DEADLETTER_OUT)
                      .build());

      // Convert the records which were successfully processed by the UDF into TableRow objects.
      PCollectionTuple jsonToTableRowOut =
          udfOut
              .get(UDF_OUT)
              .apply(
                  "JsonToTableRow",
                  FailsafeJsonToTableRow.<PubsubMessage>newBuilder()
                      .setSuccessTag(TRANSFORM_OUT)
                      .setFailureTag(TRANSFORM_DEADLETTER_OUT)
                      .build());

      // Re-wrap the PCollections so we can return a single PCollectionTuple
      return PCollectionTuple.of(UDF_OUT, udfOut.get(UDF_OUT))
          .and(UDF_DEADLETTER_OUT, udfOut.get(UDF_DEADLETTER_OUT))
          .and(TRANSFORM_OUT, jsonToTableRowOut.get(TRANSFORM_OUT))
          .and(TRANSFORM_DEADLETTER_OUT, jsonToTableRowOut.get(TRANSFORM_DEADLETTER_OUT));
    }
  }

  /**
   * The {@link PubsubMessageToFailsafeElementFn} wraps an incoming {@link PubsubMessage} with the
   * {@link FailsafeElement} class so errors can be recovered from and the original message can be
   * output to a error records table.
   */
  static class PubsubMessageToFailsafeElementFn
      extends DoFn<PubsubMessage, FailsafeElement<PubsubMessage, String>> {
    @ProcessElement
    public void processElement(ProcessContext context) {
      PubsubMessage message = context.element();
      context.output(
          FailsafeElement.of(message, new String(message.getPayload(), StandardCharsets.UTF_8)));
    }
  }
}

Python

El código fuente de Python no está disponible.

Java Database Connectivity (JDBC) a BigQuery

La plantilla de JDBC a BigQuery es una canalización por lotes que copia los datos de una tabla de base de datos relacional a una tabla de BigQuery existente. Esta canalización utiliza JDBC para conectarse a las bases de datos relacionales. Puedes usar esta plantilla para copiar datos desde cualquier base de datos relacional con controladores de JDBC disponibles en BigQuery.

Requisitos para esta canalización:

Los controladores de JDBC de la base de datos relacional deben estar disponibles.
La tabla de BigQuery debe existir antes de que se ejecute la canalización.
La tabla de BigQuery debe tener un esquema compatible.
Se debe poder acceder a la base de datos relacional desde la subred donde se ejecuta Cloud Dataflow.

Parámetros de la plantilla

Parámetro	Descripción
`driverJars`	Lista separada por comas de los archivos jar del controlador. Por ejemplo, `gs://<my-bucket>/driver_jar1.jar,gs://<my-bucket>/driver_jar2.jar`.
`driverClassName`	Nombre de la clase del controlador de JDBC. Por ejemplo, `com.mysql.jdbc.Driver`.
`connectionURL`	String de la URL de la conexión de JDBC. Por ejemplo, `jdbc:mysql://some-host:3306/sampledb`.
`query`	Consulta para ejecutar en la fuente a fin de extraer datos. Por ejemplo, `select * from sampledb.sample_table`.
`outputTable`	Ubicación de la tabla de salida de BigQuery, con el formato `<my-project>:<my-dataset>.<my-table>`.
`bigQueryLoadingTemporaryDirectory`	Directorio temporal para el proceso de carga de BigQuery. Por ejemplo, `gs://<my-bucket>/my-files/temp_dir`.
`connectionProperties`	[Opcional] String de propiedades para usar en la conexión de JDBC. Por ejemplo, `unicode=true&characterEncoding=UTF-8`.
`username`	[Opcional] Nombre de usuario para usar en la conexión de JDBC.
`password`	[Opcional] Contraseña para usar en la conexión de JDBC.

Ejecuta la plantilla de JDBC a BigQuery

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe JDBC to BigQuery templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza DRIVER_PATHS con la ruta de acceso de Cloud Storage separada por comas de los controladores de JDBC.
Reemplaza DRIVER_CLASS_NAME con el nombre de clase del controlador.
Reemplaza JDBC_CONNECTION_URL con la URL de la conexión de JDBC.
Reemplaza SOURCE_SQL_QUERY con la consulta de SQL para ejecutar en la base de datos de la fuente.
Reemplaza YOUR_DATASET con tu conjunto de datos de BigQuery y YOUR_TABLE_NAME con el nombre de tu tabla de BigQuery.
Reemplaza PATH_TO_TEMP_DIR_ON_GCS con la ruta de acceso de Cloud Storage al directorio temporal.
Reemplaza CONNECTION_PROPERTIES con las propiedades de la conexión de JDBC si es necesario.
Reemplaza CONNECTION_USERNAME con el nombre de usuario de la conexión de JDBC.
Reemplaza CONNECTION_PASSWORD con la contraseña de la conexión de JDBC.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Jdbc_to_BigQuery \
    --parameters \
driverJars=DRIVER_PATHS,\
driverClassName=DRIVER_CLASS_NAME,\
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza DRIVER_PATHS con la ruta de acceso de Cloud Storage separada por comas de los controladores de JDBC.
Reemplaza DRIVER_CLASS_NAME con el nombre de clase del controlador.
Reemplaza JDBC_CONNECTION_URL con la URL de la conexión de JDBC.
Reemplaza SOURCE_SQL_QUERY con la consulta de SQL para ejecutar en la base de datos de la fuente.
Reemplaza YOUR_DATASET con tu conjunto de datos de BigQuery y YOUR_TABLE_NAME con el nombre de tu tabla de BigQuery.
Reemplaza PATH_TO_TEMP_DIR_ON_GCS con la ruta de acceso de Cloud Storage al directorio temporal.
Reemplaza CONNECTION_PROPERTIES con las propiedades de la conexión de JDBC si es necesario.
Reemplaza CONNECTION_USERNAME con el nombre de usuario de la conexión de JDBC.
Reemplaza CONNECTION_PASSWORD con la contraseña de la conexión de JDBC.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Jdbc_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverJars": "DRIVER_PATHS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "JDBC_CONNECTION_URL",
       "query": "SOURCE_SQL_QUERY",
       "outputTable": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.io.DynamicJdbcIO;
import com.google.cloud.teleport.templates.common.JdbcConverters;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 */
public class JdbcToBigQuery {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link
   * JdbcToBigQuery#run(JdbcConverters.JdbcToBigQueryOptions)} 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) {

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

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(JdbcConverters.JdbcToBigQueryOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    pipeline
        /*
         * Step 1: Read records via JDBC and convert to TableRow
         *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
         */
        .apply(
            "Read from JdbcIO",
            DynamicJdbcIO.<TableRow>read()
                .withDataSourceConfiguration(
                    DynamicJdbcIO.DynamicDataSourceConfiguration.create(
                            options.getDriverClassName(), options.getConnectionURL())
                        .withUsername(options.getUsername())
                        .withPassword(options.getPassword())
                        .withDriverJars(options.getDriverJars())
                        .withConnectionProperties(options.getConnectionProperties()))
                .withQuery(options.getQuery())
                .withCoder(TableRowJsonCoder.of())
                .withRowMapper(JdbcConverters.getResultSetToTableRow()))
        /*
         * Step 2: Append TableRow to an existing BigQuery table
         */
        .apply(
            "Write to BigQuery",
            BigQueryIO.writeTableRows()
                .withoutValidation()
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory())
                .to(options.getOutputTable()));

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

Python

El código fuente de Python no está disponible.

Cloud Storage Text a Cloud Pub/Sub (lote)

Esta plantilla crea una canalización por lotes que lee registros desde los archivos de texto almacenados en Cloud Storage y los publica en un tema de Cloud Pub/Sub. Se puede usar para publicar registros en un archivo delimitado por saltos de línea que contiene registros JSON o un archivo CSV en un tema de Cloud Pub/Sub para el procesamiento en tiempo real. Puedes usar esta plantilla para reproducir datos en Cloud Pub/Sub.

Ten en cuenta que esta plantilla no establece una marca de tiempo en los registros individuales. Es por eso que la hora del evento será la misma que la hora de publicación durante la ejecución. Si tu canalización depende de la hora precisa del evento para el procesamiento, no debes usar esta canalización.

Requisitos para esta canalización:

Los archivos para leer deben tener el formato JSON delimitado por saltos de línea o CSV. Los registros que abarcan varias líneas en los archivos de origen pueden causar problemas de bajada, ya que cada línea dentro de los archivos se publicará como un mensaje en Cloud Pub/Sub.
El tema de Cloud Pub/Sub debe existir antes de la ejecución.

Parámetros de la plantilla

Parámetro	Descripción
`inputFilePattern`	Patrón del archivo de entrada en el que se desea leer. Por ejemplo, `gs://bucket-name/files/*.json`.
`outputTopic`	Tema de entrada de Cloud Pub/Sub en el que desea escribir. El nombre debe tener el formato `projects/<project-id>/topics/<topic-name>`.

Ejecuta la plantilla de Cloud Storage Text a Cloud Pub/Sub (lote)

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage Text to Cloud Pub/Sub (Batch) templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/files/*.json,\
outputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/files/*.json",
       "outputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * The {@code TextToPubsub} pipeline publishes records to
 * Cloud Pub/Sub from a set of files. The pipeline reads each
 * file row-by-row and publishes each record as a string message.
 * At the moment, publishing messages with attributes is unsupported.
 *
 * <p>Example Usage:
 *
 * <pre>
 * {@code mvn compile exec:java \
    -Dexec.mainClass=com.google.cloud.teleport.templates.TextToPubsub \
    -Dexec.args=" \
    --project=${PROJECT_ID} \
    --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
    --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
    --runner=DataflowRunner \
    --inputFilePattern=gs://path/to/demo_file.csv \
    --outputTopic=projects/${PROJECT_ID}/topics/${TOPIC_NAME}"
 * }
 * </pre>
 *
 */
public class TextToPubsub {

  /**
   * The custom options supported by the pipeline. Inherits
   * standard configuration options.
   */
  public interface Options extends PipelineOptions {
    @Description("The file pattern to read records from (e.g. gs://bucket/file-*.csv)")
    @Required
    ValueProvider<String> getInputFilePattern();
    void setInputFilePattern(ValueProvider<String> value);

    @Description("The name of the topic which data should be published to. "
        + "The name should be in the format of projects/<project-id>/topics/<topic-name>.")
    @Required
    ValueProvider<String> getOutputTopic();
    void setOutputTopic(ValueProvider<String> value);
  }

  /**
   * Main entry-point for the pipeline. Reads in the
   * command-line arguments, parses them, and executes
   * the pipeline.
   *
   * @param args  Arguments passed in from the command-line.
   */
  public static void main(String[] args) {

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

    run(options);
  }

  /**
   * Executes the pipeline with the provided execution
   * parameters.
   *
   * @param options The execution parameters.
   */
  public static PipelineResult run(Options options) {
    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *  1) Read from the text source.
     *  2) Write each text record to Pub/Sub
     */
    pipeline
        .apply("Read Text Data", TextIO.read().from(options.getInputFilePattern()))
        .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

Python

El código fuente de Python no está disponible.

Cloud Storage Text a Cloud Pub/Sub (transmisión)

Esta plantilla crea una canalización de transmisión que sondea continuamente los archivos de texto nuevos subidos a Cloud Storage, lee cada línea del archivo y publica strings en un tema de Cloud Pub/Sub. Publica registros en un archivo delimitado por saltos de línea que contiene registros JSON o un archivo CSV en un tema de Cloud Pub/Sub para el procesamiento en tiempo real. Puedes usar esta plantilla para reproducir datos en Cloud Pub/Sub.

Actualmente, el intervalo de sondeo es fijo y configurado en 10 segundos. Esta plantilla no establece una marca de tiempo en los registros individuales. Es por eso que la hora del evento será la misma que la hora de publicación durante la ejecución. Si tu canalización depende de la hora precisa del evento para el procesamiento, no deberías usar esta canalización.

Requisitos para esta canalización:

Los archivos de entrada deben tener el formato JSON delimitado por saltos de línea o CSV. Los registros que abarcan varias líneas en los archivos de origen pueden causar problemas de bajada, ya que cada línea dentro de los archivos se publicará como un mensaje en Cloud Pub/Sub.
El tema de Cloud Pub/Sub debe existir antes de la ejecución.
La canalización se ejecuta de forma indefinida, y deberás detenerla de forma manual.

Parámetros de la plantilla

Parámetro	Descripción
`inputFilePattern`	Patrón del archivo de entrada en el que se desea leer. Por ejemplo, `gs://bucket-name/files/*.json`.
`outputTopic`	Tema de entrada de Cloud Pub/Sub en el que desea escribir. El nombre debe tener el formato `projects/<project-id>/topics/<topic-name>`.

Ejecuta la plantilla de Cloud Storage Text a Cloud Pub/Sub (transmisión)

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage Text to Cloud Pub/Sub (Stream) templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Stream_GCS_Text_to_Cloud_PubSub

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Stream_GCS_Text_to_Cloud_PubSub \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/files/*.json,\
outputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Stream_GCS_Text_to_Cloud_PubSub

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza [YOUR_BUCKET_NAME] con el nombre del depósito de Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Stream_GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/files/*.json",
       "outputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.templates.TextToPubsub.Options;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Watch;
import org.joda.time.Duration;

/**
 * The {@code TextToPubsubStream} is a streaming version of {@code TextToPubsub} pipeline that
 * publishes records to Cloud Pub/Sub from a set of files. The pipeline continuously polls for new
 * files, reads them row-by-row and publishes each record as a string message. The polling interval
 * is fixed and equals to 10 seconds. At the moment, publishing messages with attributes is
 * unsupported.
 *
 * <p>Example Usage:
 *
 * <pre>
 * {@code mvn compile exec:java \
-Dexec.mainClass=com.google.cloud.teleport.templates.TextToPubsubStream \
-Dexec.args=" \
--project=${PROJECT_ID} \
--stagingLocation=gs://${STAGING_BUCKET}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
--tempLocation=gs://${STAGING_BUCKET}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
--runner=DataflowRunner \
--inputFilePattern=gs://path/to/*.csv \
--outputTopic=projects/${PROJECT_ID}/topics/${TOPIC_NAME}"
 * }
 * </pre>
 *
 */
public class TextToPubsubStream extends TextToPubsub {
  private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(10);

  /**
   * Main entry-point for the pipeline. Reads in the
   * command-line arguments, parses them, and executes
   * the pipeline.
   *
   * @param args  Arguments passed in from the command-line.
   */
  public static void main(String[] args) {

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

    run(options);
  }

  /**
   * Executes the pipeline with the provided execution
   * parameters.
   *
   * @param options The execution parameters.
   */
  public static PipelineResult run(Options options) {
    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *  1) Read from the text source.
     *  2) Write each text record to Pub/Sub
     */
    pipeline
      .apply(
        "Read Text Data",
        TextIO.read()
          .from(options.getInputFilePattern())
          .watchForNewFiles(DEFAULT_POLL_INTERVAL, Watch.Growth.never()))
      .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

Python

El código fuente de Python no está disponible.

Cloud Pub/Sub a Cloud Storage Text

La plantilla de Cloud Pub/Sub a Cloud Storage Text es una canalización de transmisión que lee registros desde Cloud Pub/Sub y los guarda como una serie de archivos de Cloud Storage en formato de texto. Se puede usar más adelante como una forma rápida para guardar datos en Cloud Pub/Sub. De forma predeterminada, la plantilla genera un archivo nuevo cada 5 minutos.

Requisitos para esta canalización:

El tema de Cloud Pub/Sub debe existir antes de la ejecución.
Los mensajes publicados en el tema deben tener formato de texto.
Los mensajes publicados en el tema no deben contener líneas nuevas. Ten en cuenta que el mensaje de Cloud Pub/Sub se guarda como una sola línea en el archivo de salida.

Parámetros de la plantilla

Parámetro	Descripción
`inputTopic`	Tema de Cloud Pub/Sub del que se leerá la entrada. El nombre del tema debe tener el formato `projects/<project-id>/topics/<topic-name>`.
`outputDirectory`	Ruta de acceso y prefijo del nombre de archivo para escribir los archivos de salida. Por ejemplo, `gs://bucket-name/path/`. El valor debe terminar con una barra.
`outputFilenamePrefix`	Prefijo para colocar en cada archivo con ventanas. Por ejemplo, `output-`
`outputFilenameSuffix`	Prefijo para colocar en cada archivo con ventanas, generalmente una extensión de archivo como `.txt` o `.csv`.
`outputShardTemplate`	La plantilla de fragmentación define la parte dinámica de la parte de cada archivo con ventanas. De forma predeterminada, la canalización utiliza una única fragmentación de salida para el sistema de archivo dentro de cada ventana. Esto significa que todos los datos llegarán a un único archivo por ventana. La `outputShardTemplate` es la configuración predeterminada de `W-P-SS-of-NN` en la que `W` es el período de ventana, `P` es la información del panel, `S` es el número de fragmentación y `N` es la cantidad de fragmentaciones. En el caso de un solo archivo, la parte `SS-of-NN` de `outputShardTemplate` será `00-of-01`.

Ejecuta la plantilla de Cloud Pub/Sub a Cloud Storage Text

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Selecciona the Cloud Pub/Sub to Cloud Storage Text template en el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_GCS_Text

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cloud_PubSub_to_GCS_Text \
    --parameters \
inputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME,\
outputDirectory=gs://YOUR_BUCKET_NAME/output/,\
outputFilenamePrefix=output-,\
outputFilenameSuffix=.txt

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_GCS_Text

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [YOUR_TOPIC_NAME] con el nombre de tu tema de Cloud Pub/Sub.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_PubSub_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME"
       "outputDirectory": "gs://YOUR_BUCKET_NAME/output/",
       "outputFilenamePrefix": "output-",
       "outputFilenameSuffix": ".txt",
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.io.WindowedFilenamePolicy;
import com.google.cloud.teleport.util.DurationUtils;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
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.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;

/**
 * This pipeline ingests incoming data from a Cloud Pub/Sub topic and
 * outputs the raw data into windowed files at the specified output
 * directory.
 *
 * <p> Example Usage:
 *
 * <pre>
 * mvn compile exec:java \
 -Dexec.mainClass=com.google.cloud.teleport.templates.${PIPELINE_NAME} \
 -Dexec.cleanupDaemonThreads=false \
 -Dexec.args=" \
 --project=${PROJECT_ID} \
 --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
 --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
 --runner=DataflowRunner \
 --windowDuration=2m \
 --numShards=1 \
 --inputTopic=projects/${PROJECT_ID}/topics/windowed-files \
 --outputDirectory=gs://${PROJECT_ID}/temp/ \
 --outputFilenamePrefix=windowed-file \
 --outputFilenameSuffix=.txt"
 * </pre>
 * </p>
 */
public class PubsubToText {

  /**
   * Options supported by the pipeline.
   *
   * <p>Inherits standard configuration options.</p>
   */
  public interface Options extends PipelineOptions, StreamingOptions {
    @Description("The Cloud Pub/Sub topic to read from.")
    @Required
    ValueProvider<String> getInputTopic();
    void setInputTopic(ValueProvider<String> value);

    @Description("The directory to output files to. Must end with a slash.")
    @Required
    ValueProvider<String> getOutputDirectory();
    void setOutputDirectory(ValueProvider<String> value);

    @Description("The filename prefix of the files to write to.")
    @Default.String("output")
    @Required
    ValueProvider<String> getOutputFilenamePrefix();
    void setOutputFilenamePrefix(ValueProvider<String> value);

    @Description("The suffix of the files to write.")
    @Default.String("")
    ValueProvider<String> getOutputFilenameSuffix();
    void setOutputFilenameSuffix(ValueProvider<String> value);

    @Description("The shard template of the output file. Specified as repeating sequences "
        + "of the letters 'S' or 'N' (example: SSS-NNN). These are replaced with the "
        + "shard number, or number of shards respectively")
    @Default.String("W-P-SS-of-NN")
    ValueProvider<String> getOutputShardTemplate();
    void setOutputShardTemplate(ValueProvider<String> value);

    @Description("The maximum number of output shards produced when writing.")
    @Default.Integer(1)
    Integer getNumShards();
    void setNumShards(Integer value);

    @Description("The window duration in which data will be written. Defaults to 5m. "
        + "Allowed formats are: "
        + "Ns (for seconds, example: 5s), "
        + "Nm (for minutes, example: 12m), "
        + "Nh (for hours, example: 2h).")
    @Default.String("5m")
    String getWindowDuration();
    void setWindowDuration(String value);
  }

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

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

    options.setStreaming(true);

    run(options);
  }

  /**
   * 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) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *   1) Read string messages from PubSub
     *   2) Window the messages into minute intervals specified by the executor.
     *   3) Output the windowed files to GCS
     */
    pipeline
        .apply("Read PubSub Events", PubsubIO.readStrings().fromTopic(options.getInputTopic()))
        .apply(
            options.getWindowDuration() + " Window",
            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))

        // Apply windowed file writes. Use a NestedValueProvider because the filename
        // policy requires a resourceId generated from the input value at runtime.
        .apply(
            "Write File(s)",
            TextIO.write()
                .withWindowedWrites()
                .withNumShards(options.getNumShards())
                .to(
                    new WindowedFilenamePolicy(
                        options.getOutputDirectory(),
                        options.getOutputFilenamePrefix(),
                        options.getOutputShardTemplate(),
                        options.getOutputFilenameSuffix()))
                .withTempDirectory(NestedValueProvider.of(
                    options.getOutputDirectory(),
                    (SerializableFunction<String, ResourceId>) input ->
                        FileBasedSink.convertToFileResourceIfPossible(input))));

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

Python

El código fuente de Python no está disponible.

Cloud Pub/Sub a Cloud Pub/Sub

La plantilla de Cloud Pub/Sub a Cloud Pub/Sub es una canalización de transmisión que lee mensajes desde una suscripción de Cloud Pub/Sub y los escribe en otro tema de Cloud Pub/Sub. La canalización también acepta una clave de atributo de mensaje opcional y un valor que se puede usar para filtrar los mensajes que deben escribirse en el tema de Cloud Pub/Sub. Puedes usar esta plantilla para copiar mensajes desde la suscripción de Cloud Pub/Sub en otro tema de Cloud Pub/Sub con un filtro de mensaje opcional.

Requisitos para esta canalización:

La suscripción de Cloud Pub/Sub de origen debe existir antes de la ejecución.
El tema de Cloud Pub/Sub de destino debe existir antes de la ejecución.

Parámetros de la plantilla

Parámetro	Descripción
`inputSubscription`	Suscripción de Cloud Pub/Sub del que se leerá la entrada. Por ejemplo, `projects/<project-id>/subscriptions/<subscription-name>`.
`outputTopic`	Tema de Cloud Pub/Sub en la que se escribe el resultado. Por ejemplo, `projects/<project-id>/topics/<topic-name>`.
`filterKey`	[Opcional] Eventos de filtro basados en la clave de atributo. No se aplicará ningún filtro si no se especifica `filterKey`.
`filterValue`	[Opcional] Valor del atributo del filtro para usar en caso de que se provea filterKey. Se usa un `filterValue` nulo de forma predeterminada.

Ejecuta la plantilla de Cloud Pub/Sub a Cloud Pub/Sub

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Pub/Sub to Cloud Pub/Sub templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_Cloud_PubSub

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_SUBSCRIPTION_NAME con el nombre de la subscripción Cloud Pub/Sub.
Reemplaza YOUR_TOPIC_NAME con el nombre del tema de Cloud Pub/Sub.
Reemplaza FILTER_KEY con el nombre de la clave de atributo con la que filtras mensajes.
Reemplaza FILTER_VALUE con el valor del atributo.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cloud_PubSub_to_Cloud_PubSub \
    --parameters \
inputSubscription=projects/YOUR_PROJECT_ID/subscriptions/YOUR_SUBSCRIPTION_NAME,\
outputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME,\
filterKey=FILTER_KEY,\
filterValue=FILTER_VALUE

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_Cloud_PubSub

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_SUBSCRIPTION_NAME con el nombre de la subscripción Cloud Pub/Sub.
Reemplaza YOUR_TOPIC_NAME con el nombre del tema de Cloud Pub/Sub.
Reemplaza FILTER_KEY con el nombre de la clave de atributo con la que filtras mensajes.
Reemplaza FILTER_VALUE con el valor del atributo.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_PubSub_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputSubscription": "projects/YOUR_PROJECT_ID/subscriptions/YOUR_SUBSCRIPTION_NAME",
       "outputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME",
       "filterKey": "FILTER_KEY",
       "filterValue": "FILTER_VALUE"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import java.util.Objects;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.options.Description;
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.options.Validation;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** An template that copies messages from one Pubsub subscription to another Pubsub topic. */
public class PubsubToPubsub {

  /**
   * Options supported by {@link PubsubToPubsub}.
   *
   * <p>Inherits standard configuration options.
   */
  public interface Options extends PipelineOptions, StreamingOptions {
    @Description(
        "The Cloud Pub/Sub subscription to consume from. "
            + "The name should be in the format of "
            + "projects/<project-id>/subscriptions/<subscription-name>.")
    @Validation.Required
    ValueProvider<String> getInputSubscription();

    void setInputSubscription(ValueProvider<String> inputSubscription);

    @Description(
        "The Cloud Pub/Sub topic to publish to. "
            + "The name should be in the format of "
            + "projects/<project-id>/topics/<topic-name>.")
    @Validation.Required
    ValueProvider<String> getOutputTopic();

    void setOutputTopic(ValueProvider<String> outputTopic);

    @Description(
        "Filter events based on an optional attribute key. "
            + "No filters are applied if a filterKey is not specified.")
    @Validation.Required
    ValueProvider<String> getFilterKey();

    void setFilterKey(ValueProvider<String> filterKey);

    @Description(
        "Filter attribute value to use in case a filterKey is provided. "
            + "A null filterValue is used by default.")
    @Validation.Required
    ValueProvider<String> getFilterValue();

    void setFilterValue(ValueProvider<String> filterValue);
  }

  /**
   * DoFn that will determine if events are to be filtered. If filtering is enabled, it will only
   * publish events that pass the filter else, it will publish all input events.
   */
  public static class ExtractAndFilterEventsFn extends DoFn<PubsubMessage, PubsubMessage> {

    private Boolean doFilter;
    private String inputFilterKey;
    private String inputFilterValue;
    private static final Logger LOG = LoggerFactory.getLogger(ExtractAndFilterEventsFn.class);

    @StartBundle
    public void startBundle(StartBundleContext context) {
      if (this.doFilter != null) {
        return; // Filter has been evaluated already
      }

      Options options = context.getPipelineOptions().as(Options.class);

      inputFilterKey = (options.getFilterKey() == null ? null : options.getFilterKey().get());
      inputFilterValue = (options.getFilterValue() == null ? null : options.getFilterValue().get());

      if (inputFilterKey == null) {
        this.doFilter = false;
      } else {
        this.doFilter = true;
        LOG.info(
            "Enabling event filter [key: " + inputFilterKey + "][value: " + inputFilterValue + "]");
      }
    }

    @ProcessElement
    public void processElement(ProcessContext context) {
      if (!this.doFilter) {
        // Filter is not enabled
        context.output(context.element());
      } else {
        PubsubMessage message = context.element();
        String extractedValue = message.getAttribute(this.inputFilterKey);
        if (Objects.equals(extractedValue, this.inputFilterValue)) {
          context.output(message);
        }
      }
    }
  }

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

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

    options.setStreaming(true);

    run(options);
  }

  /**
   * 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) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /**
     * Steps: 1) Read PubSubMessage with attributes from input PubSub subscription.
     *        2) Apply any filters if an attribute=value pair is provided.
     *        3) Write each PubSubMessage to output PubSub topic.
     */
    pipeline
        .apply(
            "Read PubSub Events",
            PubsubIO.readMessagesWithAttributes().fromSubscription(options.getInputSubscription()))
        .apply("Filter Events If Enabled", ParDo.of(new ExtractAndFilterEventsFn()))
        .apply("Write PubSub Events", PubsubIO.writeMessages().to(options.getOutputTopic()));

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

Python

El código fuente de Python no está disponible.

Cloud Pub/Sub a Cloud Storage Avro

La plantilla de Cloud Pub/Sub a Cloud Storage Avro es una canalización de transmisión que lee datos desde un tema de Cloud Pub/Sub y escribe archivos Avro en un depósito especificado de Cloud Storage. Esta canalización admite la duración de ventana opcional que proporciona el usuario y que se utilizará para realizar operaciones de escritura con ventanas.

Requisitos para esta canalización:

El tema de entrada de Cloud Pub/Sub debe existir antes de la ejecución de la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`inputTopic`	Tema de Cloud Pub/Sub al cual suscribirse para el consumo de mensaje. El nombre del tema debe tener el formato `projects/<project-id>/topics/<topic-name>`.
`outputDirectory`	Directorio de salida donde se archivarán los archivos Avro de salida. Agrega una / al final. Por ejemplo, `gs://example-bucket/example-directory/`.
`avroTempDirectory`	Directorio para los archivos Avro temporales. Agrega una / al final. Por ejemplo, `gs://example-bucket/example-directory/`.
`outputFilenamePrefix`	[Opcional] Prefijo de nombre de archivo de salida para los archivos Avro.
`outputFilenameSuffix`	[Opcional] Sufijo de nombre de archivo de salida para los archivos Avro.
`outputShardTemplate`	[Opcional] Plantilla de fragmentación del archivo de salida. Especificada como secuencias repetidas de letras “S” o “N” (por ejemplo: SSS-NNN). Estas se reemplazan con el número de fragmentación o la cantidad de fragmentaciones respectivamente. El formato de la plantilla predeterminada es “W-P-SS-of-NN” cuando no se especifica este parámetro.
`numShards`	[Opcional] Cantidad máxima de fragmentaciones de salida que se produce con la escritura. La cantidad máxima predeterminada de fragmentaciones es 1.
`windowDuration`	[Opcional] Duración de ventana en la que se escribirán los datos. La configuración predeterminada es de 5 m. Los formatos permitidos son: Ns (para los segundos, por ejemplo, 5 s), Nm (para los minutos, por ejemplo, 12 m) y Nh (para las horas, por ejemplo, 2 h).

Ejecuta la plantilla de Cloud Pub/Sub a Cloud Storage Avro

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Selecciona the Cloud Pub/Sub to Cloud Storage Avro template en el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_Avro

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_TOPIC_NAME con el nombre del tema de Cloud Pub/Sub.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza FILENAME_PREFIX con el prefijo del nombre del archivo de salida preferido.
Reemplaza FILENAME_SUFFIX con el sufijo del nombre del archivo de salida preferido.
Reemplaza SHARD_TEMPLATE con la plantilla de fragmentación de salida preferida.
Reemplaza NUM_SHARDS con la cantidad de fragmentaciones de salida.
Reemplaza WINDOW_DURATION con la duración de ventana de salida.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cloud_PubSub_to_Avro \
    --parameters \
inputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME,\
outputDirectory=gs://YOUR_BUCKET_NAME/output/,\
avroTempDirectory=gs://YOUR_BUCKET_NAME/temp/,\
outputFilenamePrefix=FILENAME_PREFIX,\
outputFilenameSuffix=FILENAME_SUFFIX,\
outputShardTemplate=SHARD_TEMPLATE,\
numShards=NUM_SHARDS,\
windowDuration=WINDOW_DURATION

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_PubSub_to_Avro

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_TOPIC_NAME con el nombre del tema de Cloud Pub/Sub.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza FILENAME_PREFIX con el prefijo del nombre del archivo de salida preferido.
Reemplaza FILENAME_SUFFIX con el sufijo del nombre del archivo de salida preferido.
Reemplaza SHARD_TEMPLATE con la plantilla de fragmentación de salida preferida.
Reemplaza NUM_SHARDS con la cantidad de fragmentaciones de salida.
Reemplaza WINDOW_DURATION con la duración de ventana de salida.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_PubSub_to_Avro
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME",
       "outputDirectory": "gs://YOUR_BUCKET_NAME/output/",
       "avroTempDirectory": "gs://YOUR_BUCKET_NAME/temp/",
       "outputFilenamePrefix": "FILENAME_PREFIX",
       "outputFilenameSuffix": "FILENAME_SUFFIX",
       "outputShardTemplate": "SHARD_TEMPLATE",
       "numShards": "NUM_SHARDS",
       "windowDuration": "WINDOW_DURATION"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.avro.AvroPubsubMessageRecord;
import com.google.cloud.teleport.io.WindowedFilenamePolicy;
import com.google.cloud.teleport.util.DurationUtils;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
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.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;

/**
 * This pipeline ingests incoming data from a Cloud Pub/Sub topic and outputs the raw data into
 * windowed Avro files at the specified output directory.
 *
 * <p>Files output will have the following schema:
 *
 * <pre>
 *   {
 *      "type": "record",
 *      "name": "AvroPubsubMessageRecord",
 *      "namespace": "com.google.cloud.teleport.avro",
 *      "fields": [
 *        {"name": "message", "type": {"type": "array", "items": "bytes"}},
 *        {"name": "attributes", "type": {"type": "map", "values": "string"}},
 *        {"name": "timestamp", "type": "long"}
 *      ]
 *   }
 * </pre>
 *
 * <p>Example Usage:
 *
 * <pre>
 * mvn compile exec:java \
 *   -Dexec.mainClass=com.google.cloud.teleport.templates.${PIPELINE_NAME} \
 *   -Dexec.cleanupDaemonThreads=false \
 *   -Dexec.args=" \
 *   --project=${PROJECT_ID} \
 *   --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
 *   --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
 *   --runner=DataflowRunner \
 *   --windowDuration=2m \
 *   --numShards=1 \
 *   --topic=projects/${PROJECT_ID}/topics/windowed-files \
 *   --outputDirectory=gs://${PROJECT_ID}/temp/ \
 *   --outputFilenamePrefix=windowed-file \
 *   --outputFilenameSuffix=.avro
 *   --avroTempDirectory=gs://${PROJECT_ID}/avro-temp-dir/"
 * </pre>
 */
public class PubsubToAvro {

  /**
   * Options supported by the pipeline.
   *
   * <p>Inherits standard configuration options.
   */
  public interface Options extends PipelineOptions, StreamingOptions {
    @Description("The Cloud Pub/Sub topic to read from.")
    @Required
    ValueProvider<String> getInputTopic();

    void setInputTopic(ValueProvider<String> value);

    @Description("The directory to output files to. Must end with a slash.")
    @Required
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> value);

    @Description("The filename prefix of the files to write to.")
    @Default.String("output")
    ValueProvider<String> getOutputFilenamePrefix();

    void setOutputFilenamePrefix(ValueProvider<String> value);

    @Description("The suffix of the files to write.")
    @Default.String("")
    ValueProvider<String> getOutputFilenameSuffix();

    void setOutputFilenameSuffix(ValueProvider<String> value);

    @Description(
        "The shard template of the output file. Specified as repeating sequences "
            + "of the letters 'S' or 'N' (example: SSS-NNN). These are replaced with the "
            + "shard number, or number of shards respectively")
    @Default.String("W-P-SS-of-NN")
    ValueProvider<String> getOutputShardTemplate();

    void setOutputShardTemplate(ValueProvider<String> value);

    @Description("The maximum number of output shards produced when writing.")
    @Default.Integer(1)
    Integer getNumShards();

    void setNumShards(Integer value);

    @Description(
        "The window duration in which data will be written. Defaults to 5m. "
            + "Allowed formats are: "
            + "Ns (for seconds, example: 5s), "
            + "Nm (for minutes, example: 12m), "
            + "Nh (for hours, example: 2h).")
    @Default.String("5m")
    String getWindowDuration();

    void setWindowDuration(String value);

    @Description("The Avro Write Temporary Directory. Must end with /")
    @Required
    ValueProvider<String> getAvroTempDirectory();

    void setAvroTempDirectory(ValueProvider<String> value);

  }

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

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

    run(options);
  }

  /**
   * 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) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *   1) Read messages from PubSub
     *   2) Window the messages into minute intervals specified by the executor.
     *   3) Output the windowed data into Avro files, one per window by default.
     */
    pipeline
        .apply(
            "Read PubSub Events",
            PubsubIO.readMessagesWithAttributes().fromTopic(options.getInputTopic()))
        .apply("Map to Archive", ParDo.of(new PubsubMessageToArchiveDoFn()))
        .apply(
            options.getWindowDuration() + " Window",
            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))

        // Apply windowed file writes. Use a NestedValueProvider because the filename
        // policy requires a resourceId generated from the input value at runtime.
        .apply(
            "Write File(s)",
            AvroIO.write(AvroPubsubMessageRecord.class)
                .to(
                    new WindowedFilenamePolicy(
                        options.getOutputDirectory(),
                        options.getOutputFilenamePrefix(),
                        options.getOutputShardTemplate(),
                        options.getOutputFilenameSuffix()))
                .withTempDirectory(NestedValueProvider.of(
                    options.getAvroTempDirectory(),
                    (SerializableFunction<String, ResourceId>) input ->
                        FileBasedSink.convertToFileResourceIfPossible(input)))
                /*.withTempDirectory(FileSystems.matchNewResource(
                    options.getAvroTempDirectory(),
                    Boolean.TRUE))
                    */
                .withWindowedWrites()
                .withNumShards(options.getNumShards()));

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

  /**
   * Converts an incoming {@link PubsubMessage} to the {@link AvroPubsubMessageRecord} class by
   * copying it's fields and the timestamp of the message.
   */
  static class PubsubMessageToArchiveDoFn extends DoFn<PubsubMessage, AvroPubsubMessageRecord> {
    @ProcessElement
    public void processElement(ProcessContext context) {
      PubsubMessage message = context.element();
      context.output(
          new AvroPubsubMessageRecord(
              message.getPayload(), message.getAttributeMap(), context.timestamp().getMillis()));
    }
  }
}

Python

El código fuente de Python no está disponible.

Cloud Datastore a Cloud Storage Text

La plantilla de Cloud Datastore a Cloud Storage Text es una canalización por lotes que lee entidades de Cloud Datastore y las escribe en Cloud Storage como archivos de texto. Puedes proporcionar una función para procesar cada entidad como una string JSON. Si no proporcionas esa función, cada línea en el archivo de salida será una entidad serializada JSON.

Requisitos para esta canalización:

Se debe configurar Cloud Datastore en el proyecto antes de la ejecución.

Parámetros de la plantilla

Parámetro	Descripción
`datastoreReadGqlQuery`	Consulta de GQL que especifica cuáles son las entidades que se deben tomar. Por ejemplo, `SELECT * FROM MyKind`.
`datastoreReadProjectId`	ID del proyecto de GCP de la instancia de Cloud Datastore de la que quieres leer datos.
`datastoreReadNamespace`	Espacio de nombres de las entidades solicitadas. Para usar el espacio de nombres predeterminado, deja este parámetro en blanco.
`javascriptTextTransformGcsPath`	Ruta de acceso de Cloud Storage que contiene el código de JavaScript. Por ejemplo, `gs://mybucket/mytransforms/*.js`. Si no quieres proporcionar una función, deja este parámetro en blanco.
`javascriptTextTransformFunctionName`	Nombre de la función de JavaScript para denominar. Por ejemplo, si la función de JavaScript es `function myTransform(inJson) { ...dostuff...}`, el nombre de la función es `myTransform`. Si no quieres proporcionar una función, deja este parámetro en blanco.
`textWritePrefix`	Prefijo de la ruta de acceso de Cloud Storage donde se especifican los datos que se deben escribir. Por ejemplo, `gs://mybucket/somefolder/`.

Ejecuta la plantilla de Cloud Datastore a Cloud Storage Text

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Datastore to Cloud Storage Text templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_DATASTORE_KIND con el tipo de las entidades de Datastore.
Reemplaza YOUR_DATASTORE_NAMESPACE con el espacio de nombres de las entidades de Datastore.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_GCS_Text \
    --parameters \
datastoreReadGqlQuery="SELECT * FROM YOUR_DATASTORE_KIND",\
datastoreReadProjectId=YOUR_PROJECT_ID,\
datastoreReadNamespace=YOUR_DATASTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_DATASTORE_KIND con el tipo de las entidades de Datastore.
Reemplaza YOUR_DATASTORE_NAMESPACE con el espacio de nombres de las entidades de Datastore.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Datastore_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "datastoreReadGqlQuery": "SELECT * FROM YOUR_DATASTORE_KIND"
       "datastoreReadProjectId": "YOUR_PROJECT_ID",
       "datastoreReadNamespace": "YOUR_DATASTORE_NAMESPACE",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreReadOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.ReadJsonEntities;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

/**
 * Dataflow template which copies Datastore Entities to a Text sink.
 * Text is encoded using JSON encoded entity in the v1/Entity rest format:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class DatastoreToText {

  interface DatastoreToTextOptions extends
      PipelineOptions,
      DatastoreReadOptions,
      JavascriptTextTransformerOptions,
      FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Entities from Datastore, passes in the JSON encoded Entities
   * to a Javascript UDF, and writes the JSON to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    DatastoreToTextOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(DatastoreToTextOptions.class);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(ReadJsonEntities.newBuilder()
            .setGqlQuery(options.getDatastoreReadGqlQuery())
            .setProjectId(options.getDatastoreReadProjectId())
            .setNamespace(options.getDatastoreReadNamespace())
            .build())
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(TextIO.write()
            .to(options.getTextWritePrefix())
            .withSuffix(".json"));

    pipeline.run();
  }
}

Python

El código fuente de Python no está disponible.

Cloud Storage Text a BigQuery

La canalización Cloud Storage Text a BigQuery es una canalización por lotes que te permite leer archivos de texto almacenados en Cloud Storage, transformarlos con una Función definida por el usuario de JavaScript (UDF) que proporcionas y enviar el resultado a BigQuery.

IMPORTANTE: Si vuelves a usar una tabla de BigQuery existente, se reemplazará la tabla.

Requisitos para esta canalización:

Crea un archivo JSON que describa tu esquema de BigQuery.

Asegúrate de que exista un arreglo JSON de nivel superior denominado “Esquema de BigQuery” y que su contenido siga el patrón {“name”: ‘COLUMN NAME”, “type”:”DATA TYPE”}. Por ejemplo:

{
  "BigQuery Schema": [
    {
      "name": "location",
      "type": "STRING"
    },
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "STRING"
    },
    {
      "name": "color",
      "type": "STRING"
    },
    {
      "name": "coffee",
      "type": "STRING"
    }
  ]
}

Crea un archivo de JavaScript (.js) con tu función UDF que proporciona la lógica para transformar las líneas de texto. Ten en cuenta que tu función debe mostrar una string JSON.
Por ejemplo, esta función divide cada línea de un archivo CSV y muestra una string JSON después de transformar los valores.
```
function transform(line) {
var values = line.split(',');

var obj = new Object();
obj.location = values[0];
obj.name = values[1];
obj.age = values[2];
obj.color = values[3];
obj.coffee = values[4];
var jsonString = JSON.stringify(obj);

return jsonString;
}
```

Parámetros de la plantilla

Parámetro	Descripción
`javascriptTextTransformFunctionName`	Nombre de la función que quieres llamar desde el archivo `.js`.
`JSONPath`	Ruta de acceso `gs://` al archivo JSON que define el esquema de BigQuery almacenado en Cloud Storage. Por ejemplo, `gs://path/to/my/schema.json`.
`javascriptTextTransformGcsPath`	Ruta de acceso `gs://` al archivo JavaScript que define tu UDF. Por ejemplo, `gs://path/to/my/javascript_function.js`.
`inputFilePattern`	Ruta de acceso `gs://` al texto en Cloud Storage que quieres procesar. Por ejemplo, `gs://path/to/my/text/data.txt`.
`outputTable`	Nombre de la tabla de BigQuery que quieres crear para almacenar los datos procesados. Si vuelves a usar una tabla de BigQuery existente, se reemplazará la tabla. Por ejemplo, `my-project-name:my-dataset.my-table`.
`bigQueryLoadingTemporaryDirectory`	Directorio temporal para el proceso de carga de BigQuery. Por ejemplo, `gs://my-bucket/my-files/temp_dir`.

Ejecuta la plantilla de Cloud Storage Text a BigQuery

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage Text to BigQuery templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de tu UDF.
Reemplaza PATH_TO_BIGQUERY_SCHEMA_JSON con la ruta de acceso de Cloud Storage al archivo JSON que contiene la definición del esquema.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza PATH_TO_YOUR_TEXT_DATA con la ruta de acceso de Cloud Storage a tu conjunto de datos de texto.
Reemplaza BIGQUERY_TABLE con el nombre de la tabla de BigQuery.
Reemplaza PATH_TO_TEMP_DIR_ON_GCS con la ruta de acceso de Cloud Storage al directorio temporal.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_BigQuery \
    --parameters \
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
inputFilePattern=PATH_TO_YOUR_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de tu UDF.
Reemplaza PATH_TO_BIGQUERY_SCHEMA_JSON con la ruta de acceso de Cloud Storage al archivo JSON que contiene la definición del esquema.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza PATH_TO_YOUR_TEXT_DATA con la ruta de acceso de Cloud Storage a tu conjunto de datos de texto.
Reemplaza BIGQUERY_TABLE con el nombre de la tabla de BigQuery.
Reemplaza PATH_TO_TEMP_DIR_ON_GCS con la ruta de acceso de Cloud Storage al directorio temporal.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_YOUR_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.teleport.templates.common.BigQueryConverters;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Templated pipeline to read text from TextIO, apply a javascript UDF to it, and write it to GCS.
 */
public class TextIOToBigQuery {

  /** Options supported by {@link TextIOToBigQuery}. */
  public interface Options extends DataflowPipelineOptions, JavascriptTextTransformerOptions {
    @Description("The GCS location of the text you'd like to process")
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @Description("JSON file with BigQuery Schema description")
    ValueProvider<String> getJSONPath();

    void setJSONPath(ValueProvider<String> value);

    @Description("Output topic to write to")
    ValueProvider<String> getOutputTable();

    void setOutputTable(ValueProvider<String> value);

    @Description("GCS path to javascript fn for transforming output")
    ValueProvider<String> getJavascriptTextTransformGcsPath();

    void setJavascriptTextTransformGcsPath(ValueProvider<String> jsTransformPath);

    @Validation.Required
    @Description("UDF Javascript Function Name")
    ValueProvider<String> getJavascriptTextTransformFunctionName();

    void setJavascriptTextTransformFunctionName(
        ValueProvider<String> javascriptTextTransformFunctionName);

    @Validation.Required
    @Description("Temporary directory for BigQuery loading process")
    ValueProvider<String> getBigQueryLoadingTemporaryDirectory();

    void setBigQueryLoadingTemporaryDirectory(ValueProvider<String> directory);
  }

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

  private static final String BIGQUERY_SCHEMA = "BigQuery Schema";
  private static final String NAME = "name";
  private static final String TYPE = "type";
  private static final String MODE = "mode";

  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply("Read from source", TextIO.read().from(options.getInputFilePattern()))
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(BigQueryConverters.jsonToTableRow())
        .apply(
            "Insert into Bigquery",
            BigQueryIO.writeTableRows()
                .withSchema(
                    NestedValueProvider.of(
                        options.getJSONPath(),
                        new SerializableFunction<String, TableSchema>() {

                          @Override
                          public TableSchema apply(String jsonPath) {

                            TableSchema tableSchema = new TableSchema();
                            List<TableFieldSchema> fields = new ArrayList<>();
                            SchemaParser schemaParser = new SchemaParser();
                            JSONObject jsonSchema;

                            try {

                              jsonSchema = schemaParser.parseSchema(jsonPath);

                              JSONArray bqSchemaJsonArray =
                                  jsonSchema.getJSONArray(BIGQUERY_SCHEMA);

                              for (int i = 0; i < bqSchemaJsonArray.length(); i++) {
                                JSONObject inputField = bqSchemaJsonArray.getJSONObject(i);
                                TableFieldSchema field =
                                    new TableFieldSchema()
                                        .setName(inputField.getString(NAME))
                                        .setType(inputField.getString(TYPE));

                                if (inputField.has(MODE)) {
                                  field.setMode(inputField.getString(MODE));
                                }

                                fields.add(field);
                              }
                              tableSchema.setFields(fields);

                            } catch (Exception e) {
                              throw new RuntimeException(e);
                            }
                            return tableSchema;
                          }
                        }))
                .to(options.getOutputTable())
                .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory()));

    pipeline.run();
  }
}

Python

El código fuente de Python no está disponible.

Cloud Storage Text a Cloud Datastore

La plantilla de Cloud Storage Text a Cloud Datastore es una canalización por lotes que lee archivos de texto almacenados en Cloud Storage y escribe entidades codificadas en JSON en Cloud Datastore. Cada línea en los archivos de texto de entrada debe tener especificado el formato JSON en https://cloud.google.com/datastore/docs/reference/rest/v1/Entity.

Requisitos para esta canalización:

Se debe habilitar Datastore en el proyecto de destino.

Parámetros de la plantilla

Parámetro	Descripción
`textReadPattern`	Patrón de la ruta de acceso del archivo de Cloud Storage que especifica la ubicación de tus archivos de datos de texto. Por ejemplo, `gs://mybucket/somepath/*.json`.
`javascriptTextTransformGcsPath`	Patrón de la de acceso de Cloud Storage que contiene el código de JavaScript. Por ejemplo, `gs://mybucket/mytransforms/*.js`. Si no quieres proporcionar una función, deja este parámetro en blanco.
`javascriptTextTransformFunctionName`	Nombre de la función de JavaScript para denominar. Por ejemplo, si la función de JavaScript es `function myTransform(inJson) { ...dostuff...}`, el nombre de la función es `myTransform`. Si no quieres proporcionar una función, deja este parámetro en blanco.
`datastoreWriteProjectId`	ID del proyecto de GCP donde se escriben las entidades de Cloud Datastore.
`errorWritePath`	Archivo de salida de registro de errores para escribir las fallas que ocurren durante el procesamiento. Por ejemplo, `gs://bucket-name/errors.txt`.

Ejecuta la plantilla de Cloud Storage Text a Datastore

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage Text to Datastore templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza PATH_TO_INPUT_TEXT_FILES con el patrón de archivos de entrada en Cloud Storage.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza ERROR_FILE_WRITE_PATH con la ruta de acceso deseada al archivo de error en Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Datastore \
    --parameters \
textReadPattern=PATH_TO_INPUT_TEXT_FILES,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=YOUR_PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza PATH_TO_INPUT_TEXT_FILES con el patrón de archivos de entrada en Cloud Storage.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza ERROR_FILE_WRITE_PATH con la ruta de acceso deseada al archivo de error en Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Datastore
{
   "jobName": "JOB_NAME",
   "parameters": {
       "textReadPattern": "PATH_TO_INPUT_TEXT_FILES",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "YOUR_PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.TupleTag;

/**
 * Dataflow template which reads from a Text Source and writes JSON encoded Entities into Datastore.
 * The Json is expected to be in the format of:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class TextToDatastore {

  /**
   * TextToDatastore Pipeline Options.
   */
  public interface TextToDatastoreOptions extends
      PipelineOptions,
      FilesystemReadOptions,
      JavascriptTextTransformerOptions,
      DatastoreWriteOptions,
      ErrorWriteOptions {}

  /**
   * Runs a pipeline which reads from a Text Source, passes the Text to a Javascript UDF,
   * writes the JSON encoded Entities to a TextIO sink.
   *
   * <p>If your Text Source does not contain JSON encoded Entities, then you'll need to supply a
   * Javascript UDF which transforms your data to be JSON encoded Entities.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    TextToDatastoreOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(TextToDatastoreOptions.class);

    TupleTag<String> errorTag = new TupleTag<String>("errors"){};

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(TextIO.read()
            .from(options.getTextReadPattern()))
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(WriteJsonEntities.newBuilder()
            .setProjectId(options.getDatastoreWriteProjectId())
            .setErrorTag(errorTag)
            .build())
        .apply(LogErrors.newBuilder()
            .setErrorWritePath(options.getErrorWritePath())
            .setErrorTag(errorTag)
            .build());

    pipeline.run();
  }
}

Python

El código fuente de Python no está disponible.

Cloud Storage AvroFile a Cloud Bigtable

La plantilla de Cloud Storage AvroFile a Cloud Bigtable es una canalización que lee datos de AvroFile en el depósito de Cloud Storage y los escribe en una tabla de Cloud Bigtable. Puedes usar la plantilla como una solución rápida para mover datos desde Cloud Storage hacia Cloud Bigtable.

Requisitos para esta canalización:

La tabla de Cloud Bigtable debe existir y tener las mismas familias de columnas como se exportó en AvroFile.
El AvroFile de entrada debe existir en el depósito de Cloud Storage antes de la ejecución de la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`bigtableProjectId`	ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres escribir datos.
`bigtableInstanceId`	ID de la instancia de Cloud Bigtable que contiene la tabla.
`bigtableTableId`	ID de la tabla de Cloud Bigtable para importar.
`inputFilePattern`	Patrón de la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, `gs://mybucket/somefolder/prefix*`.

Ejecuta la plantilla de Cloud Storage AvroFile a Cloud Bigtable

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage AvroFile to Cloud Bigtable templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_AvroFile_to_Cloud_Bigtable

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [INPUT_FILE_PATTERN] con la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, gs://mybucket/somefolder/prefix*.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/GCS_AvroFile_to_Cloud_Bigtable \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],inputFilePattern=[INPUT_FILE_PATTERN]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_AvroFile_to_Cloud_Bigtable

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [INPUT_FILE_PATTERN] con la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, gs://mybucket/somefolder/prefix*.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_AvroFile_to_Cloud_Bigtable
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "inputFilePattern": "[INPUT_FILE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/DataflowTemplate de GitHub.

Python

El código fuente de Python no está disponible.

Cloud Storage SequenceFile a Cloud Bigtable

La plantilla de Cloud Storage SequenceFile a Cloud Bigtable es una canalización que lee datos de SequenceFile en un depósito de Cloud Storage y los escribe en una tabla de Cloud Bigtable. Puedes usar la plantilla como una solución rápida para mover datos desde Cloud Storage hacia Cloud Bigtable.

Requisitos para esta canalización:

La tabla de Cloud Bigtable ya debe existir.
El SequenceFile de entrada debe existir en el depósito de Cloud Storage antes de la ejecución de la canalización.
El SequenceFile de entrada debe haberse exportado desde Cloud Bigtable o HBase.

Parámetros de la plantilla

Parámetro	Descripción
`bigtableProject`	ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres escribir datos.
`bigtableInstanceId`	ID de la instancia de Cloud Bigtable que contiene la tabla.
`bigtableTableId`	ID de la tabla de Cloud Bigtable para importar.
`bigtableAppProfileId`	ID del perfil de la aplicación de Cloud Bigtable que se usa para importar. Si no especificas un perfil de aplicación, Cloud Bigtable utiliza el perfil predeterminado de la aplicación de la instancia.
`sourcePattern`	Patrón de la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, `gs://mybucket/somefolder/prefix*`.

Ejecuta la plantilla de Cloud Storage SequenceFile a Cloud Bigtable

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage SequenceFile to Cloud Bigtable templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_SequenceFile_to_Cloud_Bigtable

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [APPLICATION_PROFILE_ID] con el ID del perfil de la aplicación de Cloud Bigtable que se usa para exportar.
Reemplaza [SOURCE_PATTERN] con la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, gs://mybucket/somefolder/prefix*.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/GCS_SequenceFile_to_Cloud_Bigtable \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],sourcePattern=[SOURCE_PATTERN]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_SequenceFile_to_Cloud_Bigtable

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza [PROJECT_ID] con el ID del proyecto de GCP de la instancia de Cloud Bigtable de la que quieres leer datos.
Reemplaza [INSTANCE_ID] con el ID de la instancia de Cloud Bigtable que contiene la tabla.
Reemplaza [TABLE_ID] con el ID de la tabla de Cloud Bigtable para exportar.
Reemplaza [APPLICATION_PROFILE_ID] con el ID del perfil de la aplicación de Cloud Bigtable que se usa para exportar.
Reemplaza [SOURCE_PATTERN] con la ruta de acceso de Cloud Storage donde se encuentran los datos. Por ejemplo, gs://mybucket/somefolder/prefix*.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_SequenceFile_to_Cloud_Bigtable
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "sourcePattern": "[SOURCE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/cloud-bigtable-client de GitHub.

Python

El código fuente de Python no está disponible.

Compresión masiva de archivos de Cloud Storage

La plantilla de Compresión masiva de archivos de Cloud Storage es una canalización por lotes que comprime archivos en Cloud Storage en una ubicación especificada. Esta plantilla puede ser útil cuando necesites comprimir grandes lotes de archivos como parte del proceso de archivo periódico. Los modos de compresión admitidos son BZIP2, DEFLATE, GZIP y ZIP. El resultado de los archivos en la ubicación de destino seguirá el esquema de nombres del nombre del archivo original anexado con la extensión del modo de compresión. Las extensiones anexadas serán .bzip2, .deflate, .gz y .zip.

Cualquier error que ocurra durante el proceso de compresión se enviará al archivo de falla en formato CSV del nombre de archivo, mensaje de error. Si no se ocurre ninguna falla durante la ejecución, el archivo de error se creará, pero no tendrá registros de errores.

Requisitos para esta canalización:

La compresión debe tener uno de los siguientes formatos: BZIP2, DEFLATE, GZIP y ZIP.
El directorio de salida debe existir antes de que se ejecute la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`inputFilePattern`	Patrón del archivo de entrada en el que se desea leer. Por ejemplo, `gs://bucket-name/uncompressed/*.txt`.
`outputDirectory`	Ubicación de salida en la que se desea escribir. Por ejemplo, `gs://bucket-name/compressed/`.
`outputFailureFile`	Archivo de salida del registro de errores para escribir fallas que ocurran durante el proceso de compresión. Por ejemplo, `gs://bucket-name/compressed/failed.csv`. Si no se encuentran fallas, el archivo si creará, pero estará vacío. El contenido del archivo tiene el formato CSV (nombre de archivo, error) y consta de una línea por cada archivo que tenga errores de compresión.
`compression`	Algoritmo de compresión que se utiliza para comprimir los archivos coincidentes. El modo debe ser uno de los siguientes: `BZIP2`, `DEFLATE`, `GZIP` y `ZIP`.

Ejecuta la plantilla de compresión masiva de archivos de Cloud Storage

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Bulk Compress Cloud Storage Files templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza COMPRESSION con el algoritmo de compresión que elijas.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Bulk_Compress_GCS_Files \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/uncompressed/*.txt,\
outputDirectory=gs://YOUR_BUCKET_NAME/compressed,\
outputFailureFile=gs://YOUR_BUCKET_NAME/failed/failure.csv,\
compression=COMPRESSION

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Bulk_Compress_GCS_Files

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza COMPRESSION con el algoritmo de compresión que elijas.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Bulk_Compress_GCS_Files
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/uncompressed/*.txt",
       "outputDirectory": "gs://YOUR_BUCKET_NAME/compressed",
       "outputFailureFile": "gs://YOUR_BUCKET_NAME/failed/failure.csv",
       "compression": "COMPRESSION"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.common.collect.ImmutableList;
import com.google.common.io.ByteStreams;
import java.io.IOException;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.MatchResult;
import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.util.MimeTypes;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link BulkCompressor} is a batch pipeline that compresses files on matched by an input file
 * pattern and outputs them to a specified file location. This pipeline can be useful when you need
 * to compress large batches of files as part of a perodic archival process. The supported
 * compression modes are: <code>BZIP2</code>, <code>DEFLATE</code>, <code>GZIP</code>, <code>ZIP
 * </code>. Files output to the destination location will follow a naming schema of original
 * filename appended with the compression mode extension. The extensions appended will be one of:
 * <code>.bzip2</code>, <code>.deflate</code>, <code>.gz</code>, <code>.zip</code> as determined by
 * the compression type.
 *
 * <p>Any errors which occur during the compression process will be output to the failure file in
 * CSV format of filename, error message. If no failures occur during execution, the error file will
 * still be created but will contain no error records.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>The compression must be in one of the following formats: <code>BZIP2</code>, <code>DEFLATE
 *       </code>, <code>GZIP</code>, <code>ZIP</code>.
 *   <li>The output directory must exist prior to pipeline execution.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT_ID=PROJECT ID HERE
 * PIPELINE_FOLDER=gs://${PROJECT_ID}/dataflow/pipelines/bulk-compressor
 *
 * # Set the runner
 * RUNNER=DataflowRunner
 *
 * # Build the template
 * mvn compile exec:java \
 * -Dexec.mainClass=com.google.cloud.teleport.templates.BulkCompressor \
 * -Dexec.cleanupDaemonThreads=false \
 * -Dexec.args=" \
 * --project=${PROJECT_ID} \
 * --stagingLocation=${PIPELINE_FOLDER}/staging \
 * --tempLocation=${PIPELINE_FOLDER}/temp \
 * --templateLocation=${PIPELINE_FOLDER}/template \
 * --runner=${RUNNER}"
 *
 * # Execute the template
 * JOB_NAME=bulk-compressor-$USER-`date +"%Y%m%d-%H%M%S%z"`
 *
 * gcloud dataflow jobs run ${JOB_NAME} \
 * --gcs-location=${PIPELINE_FOLDER}/template \
 * --zone=us-east1-d \
 * --parameters \
 * "inputFilePattern=${PIPELINE_FOLDER}/test/uncompressed/*,\
 * outputDirectory=${PIPELINE_FOLDER}/test/compressed,\
 * outputFailureFile=${PIPELINE_FOLDER}/test/failure/failed-${JOB_NAME}.csv,\
 * compression=GZIP"
 * </pre>
 */
public class BulkCompressor {

  /** The logger to output status messages to. */
  private static final Logger LOG = LoggerFactory.getLogger(BulkCompressor.class);

  /** The tag used to identify the main output of the {@link Compressor}. */
  private static final TupleTag<String> COMPRESOR_MAIN_OUT = new TupleTag<String>() {};

  /** The tag used to identify the dead-letter output of the {@link Compressor}. */
  private static final TupleTag<KV<String, String>> DEADLETTER_TAG =
      new TupleTag<KV<String, String>>() {};

  /**
   * The {@link Options} class provides the custom execution options passed by the executor at the
   * command-line.
   */
  public interface Options extends PipelineOptions {
    @Description("The input file pattern to read from (e.g. gs://bucket-name/uncompressed/*.gz)")
    @Required
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @Description("The output location to write to (e.g. gs://bucket-name/compressed)")
    @Required
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> value);

    @Description(
        "The output file to write failures during the compression process "
            + "(e.g. gs://bucket-name/compressed/failed.txt). The contents will be one line for "
            + "each file which failed compression. Note that this parameter will "
            + "allow the pipeline to continue processing in the event of a failure.")
    @Required
    ValueProvider<String> getOutputFailureFile();

    void setOutputFailureFile(ValueProvider<String> value);

    @Description("The compression algorithm to use on the matched files.")
    @Required
    ValueProvider<Compression> getCompression();

    void setCompression(ValueProvider<Compression> value);
  }

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * BulkCompressor#run(Options)} method to start the pipeline and invoke {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *   1) Find all files matching the input pattern
     *   2) Compress the files found and output them to the output directory
     *   3) Write any errors to the failure output file
     */
    PCollectionTuple compressOut =
        pipeline
            .apply("Match File(s)", FileIO.match().filepattern(options.getInputFilePattern()))
            .apply(
                "Compress File(s)",
                ParDo.of(new Compressor(options.getOutputDirectory(), options.getCompression()))
                    .withOutputTags(COMPRESOR_MAIN_OUT, TupleTagList.of(DEADLETTER_TAG)));

    compressOut
        .get(DEADLETTER_TAG)
        .apply(
            "Format Errors",
            MapElements.into(TypeDescriptors.strings())
                .via(kv -> String.format("%s,%s", kv.getKey(), kv.getValue())))
        .apply(
            "Write Error File",
            TextIO.write()
                .to(options.getOutputFailureFile())
                .withHeader("Filename,Error")
                .withoutSharding());

    return pipeline.run();
  }

  /**
   * The {@link Compressor} accepts {@link MatchResult.Metadata} from the FileSystems API and
   * compresses each file to an output location. Any compression failures which occur during
   * execution will be output to a separate output for further processing.
   */
  @SuppressWarnings("serial")
  public static class Compressor extends DoFn<MatchResult.Metadata, String> {

    private final ValueProvider<String> destinationLocation;
    private final ValueProvider<Compression> compressionValue;

    Compressor(ValueProvider<String> destinationLocation, ValueProvider<Compression> compression) {
      this.destinationLocation = destinationLocation;
      this.compressionValue = compression;
    }

    @ProcessElement
    public void processElement(ProcessContext context) {
      ResourceId inputFile = context.element().resourceId();
      Compression compression = compressionValue.get();

      // Add the compression extension to the output filename. Example: demo.txt -> demo.txt.gz
      String outputFilename = inputFile.getFilename() + compression.getSuggestedSuffix();

      // Resolve the necessary resources to perform the transfer
      ResourceId outputDir = FileSystems.matchNewResource(destinationLocation.get(), true);
      ResourceId outputFile =
          outputDir.resolve(outputFilename, StandardResolveOptions.RESOLVE_FILE);
      ResourceId tempFile =
          outputDir.resolve("temp-" + outputFilename, StandardResolveOptions.RESOLVE_FILE);

      // Perform the copy of the compressed channel to the destination.
      try (ReadableByteChannel readerChannel = FileSystems.open(inputFile)) {
        try (WritableByteChannel writerChannel =
            compression.writeCompressed(FileSystems.create(tempFile, MimeTypes.BINARY))) {

          // Execute the copy to the temporary file
          ByteStreams.copy(readerChannel, writerChannel);
        }

        // Rename the temporary file to the output file
        FileSystems.rename(ImmutableList.of(tempFile), ImmutableList.of(outputFile));

        // Output the path to the uncompressed file
        context.output(outputFile.toString());
      } catch (IOException e) {
        LOG.error("Error occurred during compression of {}", inputFile.toString(), e);
        context.output(DEADLETTER_TAG, KV.of(inputFile.toString(), e.getMessage()));
      }
    }
  }
}

Python

El código fuente de Python no está disponible.

Descompresión masiva de archivos de Cloud Storage

La plantilla de Descompresión masiva de archivos de Cloud Storage es una canalización por lotes que descomprime archivos en Cloud Storage en una ubicación especificada. Esta función es útil cuando deseas utilizar datos comprimidos para minimizar los costos del ancho de banda de la red durante una migración, pero quieres maximizar la velocidad de procesamiento analítico trabajando con datos no comprimidos luego de la migración. La canalización maneja de forma automática los diferentes modos de compresión durante una única ejecución y determina el modo de descompresión que utilizará en función de la extensión de archivos (.bzip2, .deflate, .gz y .zip).

Requisitos para esta canalización:

Los archivos para descomprimir deben tener uno de los siguientes formatos: Bzip2, Deflate, Gzip y Zip.
El directorio de salida debe existir antes de que se ejecute la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`inputFilePattern`	Patrón del archivo de entrada en el que se desea leer. Por ejemplo, `gs://bucket-name/compressed/*.gz`.
`outputDirectory`	Ubicación de salida en la que se desea escribir. Por ejemplo, `gs://bucket-name/decompressed`.
`outputFailureFile`	Archivo de salida del registro de errores para escribir fallas que ocurran durante el proceso de descompresión. Por ejemplo, `gs://bucket-name/decompressed/failed.csv`. Si no se encuentran fallas, el archivo si creará, pero estará vacío. El contenido del archivo tiene el formato CSV (nombre de archivo, error) y consta de una línea por cada archivo que tenga errores de descompresión.

Ejecuta la plantilla de descompresión masiva de archivos de Cloud Storage

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Bulk Decompress Cloud Storage Files templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Bulk_Decompress_GCS_Files

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza OUTPUT_FAILURE_FILE_PATH con tu ruta de acceso al archivo que contiene la información de falla.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Bulk_Decompress_GCS_Files \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/compressed/*.gz,\
outputDirectory=gs://YOUR_BUCKET_NAME/decompressed,\
outputFailureFile=OUTPUT_FAILURE_FILE_PATH

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Bulk_Decompress_GCS_Files

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza OUTPUT_FAILURE_FILE_PATH con tu ruta de acceso al archivo que contiene la información de falla.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Bulk_Decompress_GCS_Files
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/compressed/*.gz",
       "outputDirectory": "gs://YOUR_BUCKET_NAME/decompressed",
       "outputFailureFile": "OUTPUT_FAILURE_FILE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.io.ByteStreams;
import com.google.common.io.Files;
import java.io.IOException;
import java.io.StringWriter;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.MatchResult;
import org.apache.beam.sdk.io.fs.MoveOptions;
import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.util.MimeTypes;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.QuoteMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This pipeline decompresses file(s) from Google Cloud Storage and re-uploads them to a destination
 * location.
 *
 * <p><b>Parameters</b>
 *
 * <p>The {@code --inputFilePattern} parameter specifies a file glob to process. Files found can be
 * expressed in the following formats:
 *
 * <pre>
 * --inputFilePattern=gs://bucket-name/compressed-dir/*
 * --inputFilePattern=gs://bucket-name/compressed-dir/demo*.gz
 * </pre>
 *
 * <p>The {@code --outputDirectory} parameter can be expressed in the following formats:
 *
 * <pre>
 * --outputDirectory=gs://bucket-name
 * --outputDirectory=gs://bucket-name/decompressed-dir
 * </pre>
 *
 * <p>The {@code --outputFailureFile} parameter indicates the file to write the names of the files
 * which failed decompression and their associated error messages. This file can then be used for
 * subsequent processing by another process outside of Dataflow (e.g. send an email with the
 * failures, etc.). If there are no failures, the file will still be created but will be empty. The
 * failure file structure contains both the file that caused the error and the error message in CSV
 * format. The file will contain one header row and two columns (Filename, Error). The filename
 * output to the failureFile will be the full path of the file for ease of debugging.
 *
 * <pre>
 * --outputFailureFile=gs://bucket-name/decompressed-dir/failed.csv
 * </pre>
 *
 * <p>Example Output File:
 *
 * <pre>
 * Filename,Error
 * gs://docs-demo/compressedFile.gz, File is malformed or not compressed in BZIP2 format.
 * </pre>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * mvn compile exec:java \
 * -Dexec.mainClass=com.google.cloud.teleport.templates.BulkDecompressor \
 * -Dexec.cleanupDaemonThreads=false \
 * -Dexec.args=" \
 * --project=${PROJECT_ID} \
 * --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
 * --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
 * --runner=DataflowRunner \
 * --inputFilePattern=gs://${PROJECT_ID}/compressed-dir/*.gz \
 * --outputDirectory=gs://${PROJECT_ID}/decompressed-dir \
 * --outputFailureFile=gs://${PROJECT_ID}/decompressed-dir/failed.csv"
 * </pre>
 */
public class BulkDecompressor {

  /** The logger to output status messages to. */
  private static final Logger LOG = LoggerFactory.getLogger(BulkDecompressor.class);

  /**
   * A list of the {@link Compression} values excluding {@link Compression#AUTO} and {@link
   * Compression#UNCOMPRESSED}.
   */
  @VisibleForTesting
  static final Set<Compression> SUPPORTED_COMPRESSIONS =
      Stream.of(Compression.values())
          .filter(value -> value != Compression.AUTO && value != Compression.UNCOMPRESSED)
          .collect(Collectors.toSet());

  /** The error msg given when the pipeline matches a file but cannot determine the compression. */
  @VisibleForTesting
  static final String UNCOMPRESSED_ERROR_MSG =
      "Skipping file %s because it did not match any compression mode (%s)";

  @VisibleForTesting
  static final String MALFORMED_ERROR_MSG =
      "The file resource %s is malformed or not in %s compressed format.";

  /** The tag used to identify the main output of the {@link Decompress} DoFn. */
  @VisibleForTesting
  static final TupleTag<String> DECOMPRESS_MAIN_OUT_TAG = new TupleTag<String>() {};

  /** The tag used to identify the dead-letter sideOutput of the {@link Decompress} DoFn. */
  @VisibleForTesting
  static final TupleTag<KV<String, String>> DEADLETTER_TAG = new TupleTag<KV<String, String>>() {};

  /**
   * The {@link Options} class provides the custom execution options passed by the executor at the
   * command-line.
   */
  public interface Options extends PipelineOptions {
    @Description("The input file pattern to read from (e.g. gs://bucket-name/compressed/*.gz)")
    @Required
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @Description("The output location to write to (e.g. gs://bucket-name/decompressed)")
    @Required
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> value);

    @Description(
        "The output file to write failures during the decompression process "
            + "(e.g. gs://bucket-name/decompressed/failed.txt). The contents will be one line for "
            + "each file which failed decompression. Note that this parameter will "
            + "allow the pipeline to continue processing in the event of a failure.")
    @Required
    ValueProvider<String> getOutputFailureFile();

    void setOutputFailureFile(ValueProvider<String> value);
  }

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * BulkDecompressor#run(Options)} method to start the pipeline and invoke {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {

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

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {

    /*
     * Steps:
     *   1) Find all files matching the input pattern
     *   2) Decompress the files found and output them to the output directory
     *   3) Write any errors to the failure output file
     */

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    // Run the pipeline over the work items.
    PCollectionTuple decompressOut =
        pipeline
            .apply("MatchFile(s)", FileIO.match().filepattern(options.getInputFilePattern()))
            .apply(
                "DecompressFile(s)",
                ParDo.of(new Decompress(options.getOutputDirectory()))
                    .withOutputTags(DECOMPRESS_MAIN_OUT_TAG, TupleTagList.of(DEADLETTER_TAG)));

    decompressOut
        .get(DEADLETTER_TAG)
        .apply(
            "FormatErrors",
            MapElements.into(TypeDescriptors.strings())
                .via(
                    kv -> {
                      StringWriter stringWriter = new StringWriter();
                      try {
                        CSVPrinter printer =
                            new CSVPrinter(
                                stringWriter,
                                CSVFormat.DEFAULT
                                    .withQuoteMode(QuoteMode.NONE)
                                    .withRecordSeparator('\n')
                                    .withEscape('\\'));
                        printer.printRecord(kv.getKey(), kv.getValue());
                      } catch (IOException e) {
                        throw new RuntimeException(e);
                      }

                      return stringWriter.toString();
                    }))

        // We don't expect error files to be large so we'll create a single
        // file for ease of reprocessing by processes outside of Dataflow.
        .apply(
            "WriteErrorFile",
            TextIO.write()
                .to(options.getOutputFailureFile())
                .withHeader("Filename,Error")
                .withoutSharding());

    return pipeline.run();
  }

  /**
   * Performs the decompression of an object on Google Cloud Storage and uploads the decompressed
   * object back to a specified destination location.
   */
  @SuppressWarnings("serial")
  public static class Decompress extends DoFn<MatchResult.Metadata, String> {

    private final ValueProvider<String> destinationLocation;

    Decompress(ValueProvider<String> destinationLocation) {
      this.destinationLocation = destinationLocation;
    }

    @ProcessElement
    public void processElement(ProcessContext context) {
      ResourceId inputFile = context.element().resourceId();

      // Output a record to the failure file if the file doesn't match a known compression.
      if (!Compression.AUTO.isCompressed(inputFile.toString())) {
        String errorMsg =
            String.format(UNCOMPRESSED_ERROR_MSG, inputFile.toString(), SUPPORTED_COMPRESSIONS);

        context.output(DEADLETTER_TAG, KV.of(inputFile.toString(), errorMsg));
      } else {
        try {
          ResourceId outputFile = decompress(inputFile);
          context.output(outputFile.toString());
        } catch (IOException e) {
          LOG.error(e.getMessage());
          context.output(DEADLETTER_TAG, KV.of(inputFile.toString(), e.getMessage()));
        }
      }
    }

    /**
     * Decompresses the inputFile using the specified compression and outputs to the main output of
     * the {@link Decompress} doFn. Files output to the destination will be first written as temp
     * files with a "temp-" prefix within the output directory. If a file fails decompression, the
     * filename and the associated error will be output to the dead-letter.
     *
     * @param inputFile The inputFile to decompress.
     * @return A {@link ResourceId} which points to the resulting file from the decompression.
     */
    private ResourceId decompress(ResourceId inputFile) throws IOException {
      // Remove the compressed extension from the file. Example: demo.txt.gz -> demo.txt
      String outputFilename = Files.getNameWithoutExtension(inputFile.toString());

      // Resolve the necessary resources to perform the transfer.
      ResourceId outputDir = FileSystems.matchNewResource(destinationLocation.get(), true);
      ResourceId outputFile =
          outputDir.resolve(outputFilename, StandardResolveOptions.RESOLVE_FILE);
      ResourceId tempFile =
          outputDir.resolve(Files.getFileExtension(inputFile.toString())
              + "-temp-" + outputFilename, StandardResolveOptions.RESOLVE_FILE);

      // Resolve the compression
      Compression compression = Compression.detect(inputFile.toString());

      // Perform the copy of the decompressed channel into the destination.
      try (ReadableByteChannel readerChannel =
          compression.readDecompressed(FileSystems.open(inputFile))) {
        try (WritableByteChannel writerChannel = FileSystems.create(tempFile, MimeTypes.TEXT)) {
          ByteStreams.copy(readerChannel, writerChannel);
        }

        // Rename the temp file to the output file.
        FileSystems.rename(
            ImmutableList.of(tempFile),
            ImmutableList.of(outputFile),
            MoveOptions.StandardMoveOptions.IGNORE_MISSING_FILES);
      } catch (IOException e) {
        String msg = e.getMessage();

        LOG.error("Error occurred during decompression of {}", inputFile.toString(), e);
        throw new IOException(sanitizeDecompressionErrorMsg(msg, inputFile, compression));
      }

      return outputFile;
    }

    /**
     * The error messages coming from the compression library are not consistent across compression
     * modes. Here we'll attempt to unify the messages to inform the user more clearly when we've
     * encountered a file which is not compressed or malformed. Note that GZIP and ZIP compression
     * modes will not throw an exception when a decompression is attempted on a file which is not
     * compressed.
     *
     * @param errorMsg The error message thrown during decompression.
     * @param inputFile The input file which failed decompression.
     * @param compression The compression mode used during decompression.
     * @return The sanitized error message. If the error was not from a malformed file, the same
     *     error message passed will be returned.
     */
    private String sanitizeDecompressionErrorMsg(
        String errorMsg, ResourceId inputFile, Compression compression) {
      if (errorMsg != null
          && (errorMsg.contains("not in the BZip2 format")
              || errorMsg.contains("incorrect header check"))) {
        errorMsg = String.format(MALFORMED_ERROR_MSG, inputFile.toString(), compression);
      }

      return errorMsg;
    }
  }
}

Python

El código fuente de Python no está disponible.

Borrado masivo de Cloud Datastore

La plantilla de Borrado masivo de Cloud Datastore es una canalización que realiza operaciones de lectura en entidades de Datastore con una consulta de GQL específica y, luego, borra todas las entidades coincidentes en el proyecto de destino seleccionado. De forma opcional, la canalización puede pasar las entidades de Datastore codificadas en JSON a tu UDF de JavaScript, la que puedes usar para filtrar entidades mostrando valores nulos.

Requisitos para esta canalización:

Se debe configurar Cloud Datastore en el proyecto antes de la ejecución.
Si se realiza la lectura y el borrado desde instancias separadas de Datastore, la Cuenta de servicio del controlador de Dataflow debe tener permisos para escribir desde una instancia y borrar desde otra.

Parámetros de la plantilla

Parámetro	Descripción
`datastoreReadGqlQuery`	Consulta de GQL que especifica las entidades que deben coincidir para borrar. Por ejemplo, "SELECT * FROM MyKind".
`datastoreReadProjectId`	ID del proyecto de GCP de la instancia de Cloud Datastore de la que quieres leer entidades (con la consulta de GQL) que se usan para las coincidencias.
`datastoreDeleteProjectId`	ID del proyecto de GCP de la instancia de Cloud Datastore de la que quieres borrar entidades coincidentes. Esto puede ser igual a `datastoreReadProjectId` si quieres leer y borrar dentro de la misma instancia de Cloud Datastore.
`datastoreReadNamespace`	[Opcional] Espacio de nombres de las entidades solicitadas. Configurado como "" para el espacio de nombres predeterminado.
`javascriptTextTransformGcsPath`	[Opcional] Ruta de acceso de Cloud Storage que contiene el código de JavaScript. Por ejemplo, "gs://mybucket/mytransforms/*.js". Si no quieres usar una UDF, deja este campo en blanco.
`javascriptTextTransformFunctionName`	[Opcional] Nombre de la función para llamar. Si esta función muestra un valor no definido o nulo para una entidad de Datastore determinada, esa entidad no se borrará. Si tienes el código de JavaScript "function myTransform(inJson) { …dostuff…}", el nombre de la función será "myTransform". Si no quieres usar una UDF, deja este campo en blanco.

Ejecuta la plantilla de descompresión masiva de archivos de Cloud Storage

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Datastore Bulk Delete templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo). El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Bulk_Decompress_GCS_Files

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza GQL_QUERY con la consulta que usarás para borrar las entidades coincidentes.
Reemplaza DATASTORE_READ_AND_DELETE_PROJECT_ID con tu ID del proyecto de la instancia de Datastore. Este ejemplo leerá y borrará desde la misma instancia de Datastore.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_Datastore_Delete \
    --parameters \
datastoreReadGqlQuery="GQL_QUERY",\
datastoreReadProjectId=DATASTORE_READ_AND_DELETE_PROJECT_ID,\
datastoreDeleteProjectId=DATASTORE_READ_AND_DELETE_PROJECT_ID

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Bulk_Decompress_GCS_Files

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza GQL_QUERY con la consulta que usarás para borrar las entidades coincidentes.
Reemplaza DATASTORE_READ_AND_DELETE_PROJECT_ID con tu ID del proyecto de la instancia de Datastore. Este ejemplo leerá y borrará desde la misma instancia de Datastore.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Datastore_to_Datastore_Delete
{
   "jobName": "JOB_NAME",
   "parameters": {
       "datastoreReadGqlQuery": "GQL_QUERY",
       "datastoreReadProjectId": "READ_PROJECT_ID",
       "datastoreDeleteProjectId": "DELETE_PROJECT_ID"
   },
   "environment": { "zone": "us-central1-f" }
   }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreDeleteEntityJson;
import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreDeleteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreReadOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.ReadJsonEntities;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

/**
 * Dataflow template which deletes pulled Datastore Entities.
 */
public class DatastoreToDatastoreDelete {
  interface DatastoreToDatastoreDeleteOptions extends
      PipelineOptions,
      DatastoreReadOptions,
      JavascriptTextTransformerOptions,
      DatastoreDeleteOptions {}

  /**
   * Runs a pipeline which reads in Entities from datastore, passes in the JSON encoded Entities
   * to a Javascript UDF, and deletes all the Entities.
   *
   * <p>If the UDF returns value of undefined or null for a given Entity, then that Entity will not
   * be deleted.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    DatastoreToDatastoreDeleteOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(DatastoreToDatastoreDeleteOptions.class);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(ReadJsonEntities.newBuilder()
            .setGqlQuery(options.getDatastoreReadGqlQuery())
            .setProjectId(options.getDatastoreReadProjectId())
            .setNamespace(options.getDatastoreReadNamespace())
            .build())
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(DatastoreDeleteEntityJson.newBuilder()
            .setProjectId(options.getDatastoreDeleteProjectId())
            .build());

    pipeline.run();
  }

}

Python

El código fuente de Python no está disponible.

Cloud Spanner a Cloud Storage Avro

La plantilla de Cloud Spanner a Cloud Storage Avro es una canalización por lotes que exporta bases de datos completas de Cloud Spanner a Cloud Storage en formato Avro. Exportar una base de datos de Cloud Spanner crea una carpeta en el depósito que selecciones. La carpeta contiene estos archivos:

Un archivo spanner-export.json
Un archivo TableName-manifest.json para cada tabla en la base de datos que exportaste
Uno o más archivos TableName.avro-#####-of-#####

Por ejemplo, exportar una base de datos con dos tablas, Singers y Albums, crea el siguiente conjunto de archivos:

Albums-manifest.json
Albums.avro-00000-of-00002
Albums.avro-00001-of-00002
Singers-manifest.json
Singers.avro-00000-of-00003
Singers.avro-00001-of-00003
Singers.avro-00002-of-00003
spanner-export.json

Requisitos para esta canalización:

La base de datos de Cloud Spanner debe existir.
El depósito de salida de Cloud Storage debe existir.
Además de las funciones de Cloud IAM necesarias para ejecutar los trabajos de Cloud Dataflow, debes tener las funciones de Cloud IAM apropiadas a fin de leer tus datos de Cloud Spanner y escribir en tu depósito de Cloud Storage.

Parámetros de la plantilla

Parámetro	Descripción
`instanceId`	ID de instancia de la base de datos de Cloud Spanner que quieres exportar.
`databaseId`	ID de la base de datos de la base de datos de Cloud Spanner que quieres exportar.
`outputDir`	Ruta de acceso de Cloud Storage desde la que quieres exportar los archivos Avro. El trabajo de exportación crea un directorio nuevo en esta ruta de acceso que contiene los archivos exportados.

Ejecuta la plantilla

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Spanner to Cloud Storage Avro templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo).
- El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
- El nombre del trabajo debe coincidir con el formato cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] para que aparezca en la porción de Cloud Spanner de GCP Console.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

Reemplaza [DATAFLOW_REGION] con la región donde quieres ejecutar el trabajo de Cloud Dataflow (como us-central1).
Reemplaza [YOUR_INSTANCE_ID] con el ID de la instancia de Cloud Spanner.
Reemplaza [YOUR_DATABASE_ID] con el ID de la base de datos de Cloud Spanner.
Reemplaza [YOUR_GCS_DIRECTORY] con la ruta de acceso de Cloud Storage desde la que se deben exportar los archivos Avro,.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo.
- El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
- El nombre del trabajo debe coincidir con el formato cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] para que aparezca en la porción de Cloud Spanner de GCP Console.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],outputDir=[YOUR_GCS_DIRECTORY]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

Reemplaza [YOUR_PROJECT_ID] con el ID del proyecto.
Reemplaza [DATAFLOW_REGION] con la región donde quieres ejecutar el trabajo de Cloud Dataflow (como us-central1).
Reemplaza [YOUR_INSTANCE_ID] con el ID de la instancia de Cloud Spanner.
Reemplaza [YOUR_DATABASE_ID] con el ID de la base de datos de Cloud Spanner.
Reemplaza [YOUR_GCS_DIRECTORY] con la ruta de acceso de Cloud Storage desde la que se deben exportar los archivos Avro,.
Reemplaza [JOB_NAME] con un nombre a elección para el trabajo.
- El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
- El nombre del trabajo debe coincidir con el formato cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] para que aparezca en la porción de Cloud Spanner de GCP Console.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "outputDir": "gs://[YOUR_GCS_DIRECTORY]"
   }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/DataflowTemplates de GitHub.

Python

El código fuente de Python no está disponible.

Cloud Spanner a Cloud Storage Text

La plantilla de Cloud Spanner a Cloud Storage Text es una canalización por lotes que lee datos desde una tabla de Cloud Spanner, puede transformar los datos mediante una Función definida por el usuario de JavaScript (UDF) que proporcionas y escribe en Cloud Storage como archivos de texto CSV.

Requisitos para esta canalización:

La tabla de entrada Cloud Spanner debe existir antes de la ejecución de la canalización.

Parámetros de la plantilla

Parámetro	Descripción
`spannerProjectId`	ID del proyecto de GCP de la base de datos de Cloud Spanner desde la que quieres leer datos.
`spannerDatabaseId`	Base de datos de la tabla solicitada.
`spannerInstanceId`	Instancia de la tabla solicitada.
`spannerTable`	Tabla para exportar.
`textWritePrefix`	Directorio de salida donde se escribirán los archivos de texto de salida. Agrega una / al final. Por ejemplo, `gs://mybucket/somefolder/`.
`javascriptTextTransformGcsPath`	[Opcional] Ruta de acceso de Cloud Storage que contiene el código de JavaScript. Por ejemplo, `gs://mybucket/mytransforms/*.js`. Si no quieres proporcionar una función, deja este parámetro en blanco.
`javascriptTextTransformFunctionName`	[Opcional] Nombre de la función de JavaScript para llamar. Por ejemplo, si la función de JavaScript es `function myTransform(inJson) { ...dostuff...}`, el nombre de la función es `myTransform`. Si no quieres proporcionar una función, deja este parámetro en blanco.

Ejecuta la plantilla de Cloud Spanner a Cloud Storage Text

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Spanner to Cloud Storage Text templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo).
- El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
- El nombre del trabajo debe coincidir con el formato cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] para que aparezca en la porción de Cloud Spanner de GCP Console.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_DATABASE_ID con el ID de la base de datos de Cloud Spanner.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_INSTANCE_ID con el ID de la instancia de Cloud Spanner.
Reemplaza YOUR_TABLE_ID con el ID de la tabla de Cloud Spanner.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters \
spannerProjectId=YOUR_PROJECT_ID,\
spannerDatabaseId=YOUR_DATABASE_ID,\
spannerInstanceId=YOUR_INSTANCE_ID,\
spannerTable=YOUR_TABLE_ID,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

Para ejecutar esta plantilla con una solicitud de la API de REST, envía una solicitud HTTP POST con tu ID del proyecto. Esta solicitud requiere de una autorización.

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_DATABASE_ID con el ID de la base de datos de Cloud Spanner.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_INSTANCE_ID con el ID de la instancia de Cloud Spanner.
Reemplaza YOUR_TABLE_ID con el ID de la tabla de Cloud Spanner.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "spannerProjectId": "YOUR_PROJECT_ID",
       "spannerDatabaseId": "YOUR_DATABASE_ID",
       "spannerInstanceId": "YOUR_INSTANCE_ID",
       "spannerTable": "YOUR_TABLE_ID",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION"
   },
   "environment": { "zone": "us-central1-f" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2018 Google Inc.
 *
 * 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.templates;

import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.SpannerConverters;
import com.google.cloud.teleport.templates.common.SpannerConverters.SpannerReadOptions;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which copies a Spanner table to a Text sink. It exports a Spanner table using
 * <a href="https://cloud.google.com/spanner/docs/reads#read_data_in_parallel">Batch API</a>,
 * which creates multiple workers in parallel for better performance. The result is written
 * to a CSV file in Google Cloud Storage. The table schema file is saved in json format along with
 * the exported table.
 *
 * <p>Schema file sample: { "id":"INT64", "name":"STRING(MAX)" }
 *
 * <p>A sample run:
 *
 * <pre>
 * mvn compile exec:java \
 *   -Dexec.mainClass=com.google.cloud.teleport.templates.SpannerToText \
 *   -Dexec.args="--runner=DataflowRunner \
 *                --spannerProjectId=projectId
 *                --gcpTempLocation=gs://gsTmpLocation \
 *                --spannerInstanceId=instanceId \
 *                --spannerDatabaseId=databaseId \
 *                --spannerTable=table_name \
 *                --textWritePrefix=gcsOutputPath"
 * </pre>
 */
public class SpannerToText {

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

  private interface SpannerToTextOptions
      extends PipelineOptions,
          SpannerReadOptions,
          JavascriptTextTransformerOptions,
          FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Records from Spanner, passes in the CSV records to a Javascript
   * UDF, and writes the CSV to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    LOG.info("Starting pipeline setup");
    PipelineOptionsFactory.register(SpannerToTextOptions.class);
    SpannerToTextOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SpannerToTextOptions.class);

    FileSystems.setDefaultPipelineOptions(options);
    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(options.getSpannerProjectId())
            .withInstanceId(options.getSpannerInstanceId())
            .withDatabaseId(options.getSpannerDatabaseId());

    PTransform<PBegin, PCollection<ReadOperation>> spannerExport =
        SpannerConverters.ExportTransformFactory.create(
            options.getSpannerTable(), spannerConfig, options.getTextWritePrefix());

    PCollection<String> csv =
        pipeline
            .apply("Create export", spannerExport)
            // We need to use SpannerIO.readAll() instead of SpannerIO.read()
            // because ValueProvider parameters such as table name required for SpannerIO.read()
            // can be read only inside DoFn but SpannerIO.read() is of type
            // PTransform<PBegin, Struct>, which prevents prepending it with DoFn that reads these
            // parameters at the pipeline execution time.
            .apply("Read all records", SpannerIO.readAll().withSpannerConfig(spannerConfig))
            .apply(
                "Struct To Csv",
                MapElements.into(TypeDescriptors.strings())
                    .via(struct -> (new SpannerConverters.StructCsvPrinter()).print(struct)));

    if (options.getJavascriptTextTransformGcsPath().isAccessible()) {
      // The UDF function takes a CSV row as an input and produces a transformed CSV row
      csv =
          csv.apply(
              "JavascriptUDF",
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }
    csv.apply(
        "Write to storage", TextIO.write().to(options.getTextWritePrefix()).withSuffix(".csv"));

    pipeline.run();
    LOG.info("Completed pipeline setup");
  }
}

Python

El código fuente de Python no está disponible.

Cloud Storage Avro a Cloud Spanner

La plantilla de Cloud Storage Avro a Cloud Spanner es una canalización por lotes que lee archivos Avro desde Cloud Storage y los importa en una base de datos de Cloud Storage.

Requisitos para esta canalización:

La base de datos de destino de Cloud Spanner debe existir y estar vacía.
Debes tener los permisos de lectura en el depósito de Cloud Storage y los de escritura en la base de datos de destino de Cloud Spanner.
La ruta de acceso de entrada de Cloud Storage debe existir y debe incluir un archivo spanner-export.json que contenga una descripción de los archivos JSON para importar.

Parámetros de la plantilla

Parámetro	Descripción
`instanceId`	ID de la instancia de la base de datos de Cloud Spanner.
`databaseId`	ID de la base de datos de la base de datos de Cloud Spanner.
`inputDir`	Ruta de acceso de Cloud Storage desde donde se deben exportar los archivos Avro.

Ejecuta la plantilla

CONSOLE

Ejecuta desde Google Cloud Platform Console

Ve a la página de Cloud Dataflow en GCP Console.

Ir a la página de Cloud Dataflow

Haz clic en CREATE JOB FROM TEMPLATE (CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA).

Seleccionathe Cloud Storage Avro to Cloud Spanner templateen el menú desplegable Cloud Dataflow template (Plantilla de Cloud Dataflow).
Ingresa un nombre para el trabajo en el campo Job Name (Nombre del trabajo).
- El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
- El nombre del trabajo debe coincidir con el formato cloud-spanner-import-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] para que aparezca en la porción de Cloud Spanner de GCP Console.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Run Job (Ejecutar trabajo).

GCLOUD

Ejecuta desde la herramienta de línea de comandos de gcloud

Nota: Para ejecutar las plantillas a través de la herramienta de línea de comandos de gcloud, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_DATABASE_ID con el ID de la base de datos de Cloud Spanner.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_INSTANCE_ID con el ID de la instancia de Cloud Spanner.
Reemplaza YOUR_TABLE_ID con el ID de la tabla de Cloud Spanner.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/GCS_Avro_to_Cloud_Spanner' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],inputDir=[YOUR_GCS_DIRECTORY]

API

Ejecuta desde la API de REST

Para ejecutar esta plantilla, necesitarás la ruta de acceso de la plantilla en Cloud Storage:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID con el ID del proyecto.
Reemplaza JOB_NAME con un nombre a elección para el trabajo. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para que sea válido.
Reemplaza YOUR_DATABASE_ID con el ID de la base de datos de Cloud Spanner.
Reemplaza YOUR_BUCKET_NAME con el nombre del depósito de Cloud Storage.
Reemplaza YOUR_INSTANCE_ID con el ID de la instancia de Cloud Spanner.
Reemplaza YOUR_TABLE_ID con el ID de la tabla de Cloud Spanner.
Reemplaza PATH_TO_JAVASCRIPT_UDF_FILE con la ruta de acceso de Cloud Storage al archivo .js que contiene el código de JavaScript.
Reemplaza YOUR_JAVASCRIPT_FUNCTION con el nombre de la función de JavaScript.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "inputDir": "gs://[YOUR_GCS_DIRECTORY]"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

Código fuente de la plantilla

Java: SDK 2.x

El código fuente de esta plantilla se encuentra en el repositorio GoogleCloudPlatform/DataflowTemplates de GitHub.

Python

El código fuente de Python no está disponible.