Comienza con las plantillas proporcionadas por Google

Google proporciona un conjunto de plantillas de código abierto de Dataflow. Para obtener información general sobre las plantillas, consulta la página Descripción general. Para comenzar, usa la plantilla WordCount. Observa otras plantillas proporcionadas por Google:

Plantillas de transmisión

Plantillas para procesar datos de forma continua:

• Suscripción de Pub/Sub a BigQuery
• Tema de Pub/Sub a BigQuery
• Pub/Sub Avro a BigQuery
• Proto de Pub/Sub a BigQuery
• Pub/Sub a Pub/Sub
• Pub/Sub a Splunk
• Pub/Sub a Cloud Storage Avro
• Pub/Sub a Cloud Storage Text
• Pub/Sub a MongoDB
• Pub/Sub a Elasticsearch
• Cloud Storage Text en BigQuery (transmisión)
• Flujos de cambio de Cloud Spanner a BigQuery
• Flujos de cambio de Cloud Spanner a Cloud Storage
• Cloud Storage Text a Pub/Sub (transmisión)
• Enmascaramiento de datos y asignación de tokens de Cloud Storage a BigQuery (con Cloud DLP)
• Captura de datos modificados a BigQuery (transmisión)
• Apache Kafka a BigQuery
• Transmisión de datos a BigQuery
• Datastream a PostgreSQL
• Datastream a Cloud Spanner
• Conectividad a bases de datos de Java (JDBC) de Pub/Sub

Plantillas de lotes

Plantillas para procesar datos de forma masiva:

• BigQuery a TFRecord de Cloud Storage
• Exportación de BigQuery a Parquet (a través de la API de Storage)
• BigQuery a Elasticsearch
• Bigtable a Avro en Cloud Storage
• Bigtable a Parquet en Cloud Storage
• Bigtable a archivos SequenceFile de Cloud Storage
• Datastore a Cloud Storage en archivos de texto [obsoleta]
• Firestore (modo Datastore) a Cloud Storage Text
• Cloud Spanner a Cloud Storage Avro
• Cloud Spanner a Cloud Storage Text
• Avro de Cloud Storage a Bigtable
• Cloud Storage Avro a Cloud Spanner
• Parquet de Cloud Storage a Bigtable
• SequenceFiles de Cloud Storage a Bigtable
• Cloud Storage Text a BigQuery
• Cloud Storage en archivos de texto a Datastore [obsoleta]
• Cloud Storage Text a Firestore (modo Datastore)
• Cloud Storage Text a Pub/Sub (por lotes)
• Cloud Storage Text a Cloud Spanner
• Cloud Storage a Elasticsearch
• Conectividad a bases de datos de Java (JDBC) para BigQuery
• Conectividad a bases de datos de Java (JDBC) para Pub/Sub
• Apache Cassandra a Bigtable

Plantillas de utilidad

• Conversión de formato de archivo
• Compresión masiva de archivos de Cloud Storage
• Descompresión masiva de archivos de Cloud Storage
• Eliminación masiva de Datastore [obsoleto]
• Eliminación masiva de Firestore (modo Datastore)
• Generador de datos de transmisión a Pub/Sub, BigQuery y Cloud Storage

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.

Si el bucket de Cloud Storage está fuera de tu perímetro de servicio, crea unaregla de salida que permita el acceso al bucket

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

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

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

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

gcloud

Nota: Para usar Google Cloud CLI a fin de ejecutar plantillas classic, debes tener Google Cloud CLI versión 138.0.0 o una versión posterior.

En tu shell o terminal, ejecuta la plantilla:

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

Reemplaza lo siguiente:

• JOB_NAME: Es el nombre del trabajo que elijas
• REGION_NAME: El extremo regional en el que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.

API

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

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

Reemplaza lo siguiente:

• PROJECT_ID: El ID del proyecto de Cloud en el que deseas ejecutar el trabajo de Dataflow.
• JOB_NAME: Es el nombre del trabajo que elijas
• LOCATION: El extremo regional en el que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.

Código fuente de la plantilla

Java

Ver en GitHubComentarios

/*
 * Copyright (C) 2016 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.templates;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
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.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;

/** An template that counts words in Shakespeare. */
public class WordCount {

  static class ExtractWordsFn extends DoFn<String, String> {
    private final Counter emptyLines = Metrics.counter(ExtractWordsFn.class, "emptyLines");

    @ProcessElement
    public void processElement(ProcessContext c) {
      if (c.element().trim().isEmpty()) {
        emptyLines.inc();
      }

      // Split the line into words.
      String[] words = c.element().split("[^a-zA-Z']+");

      // Output each word encountered into the output PCollection.
      for (String word : words) {
        if (!word.isEmpty()) {
          c.output(word);
        }
      }
    }
  }

  /** A SimpleFunction that converts a Word and Count into a printable string. */
  public static class FormatAsTextFn extends SimpleFunction<KV<String, Long>, String> {
    @Override
    public String apply(KV<String, Long> input) {
      return input.getKey() + ": " + input.getValue();
    }
  }

  /**
   * A PTransform that converts a PCollection containing lines of text into a PCollection of
   * formatted word counts.
   */
  public static class CountWords
      extends PTransform<PCollection<String>, PCollection<KV<String, Long>>> {
    @Override
    public PCollection<KV<String, Long>> expand(PCollection<String> lines) {

      // Convert lines of text into individual words.
      PCollection<String> words = lines.apply(ParDo.of(new ExtractWordsFn()));

      // Count the number of times each word occurs.
      PCollection<KV<String, Long>> wordCounts = words.apply(Count.<String>perElement());

      return wordCounts;
    }
  }

  /**
   * Options supported by {@link com.google.cloud.teleport.templates.WordCount}.
   *
   * <p>Inherits standard configuration options.
   */
  public interface WordCountOptions extends PipelineOptions {
    @Description("Path of the file to read from")
    ValueProvider<String> getInputFile();

    void setInputFile(ValueProvider<String> value);

    @Description("Path of the file to write to")
    ValueProvider<String> getOutput();

    void setOutput(ValueProvider<String> value);
  }

  public static void main(String[] args) {
    WordCountOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(WordCountOptions.class);
    Pipeline p = Pipeline.create(options);
    p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
        .apply(new CountWords())
        .apply(MapElements.via(new FormatAsTextFn()))
        .apply("WriteCounts", TextIO.write().to(options.getOutput()));

    p.run();
  }
}

Extender plantillas

Puedes compilar tus propias canalizaciones si extiendes las plantillas de Dataflow de código abierto. Por ejemplo, en el caso de una plantilla que usa una duración de ventana fija, los datos que llegan fuera de la ventana podrían descartarse. A fin de evitar este comportamiento, usa el código de la plantilla como base y modifícalo para invocar la operación .withAllowedLateness.