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 de WordCount documentada en la sección a continuación. Observa otras plantillas proporcionadas por Google:

Plantillas de transmisión: plantillas para procesar datos de forma continua:

Plantillas por lotes: plantillas para procesar datos de forma masiva:

Plantillas de utilidad:

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 Console

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

Ir a la página de Dataflow

Haz clic en Crear trabajo a partir de una plantilla (Create job from template).

Selecciona the WordCount template en el menú desplegable Plantilla de Dataflow.
Ingresa un nombre para el trabajo en el campo 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 ser 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 usar la herramienta de línea de comandos de gcloud a fin de ejecutar plantillas, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Cuando ejecutes esta plantilla, necesitarás la ruta de Cloud Storage a la plantilla:

gs://dataflow-templates/latest/Word_Count

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza JOB_NAME por un nombre de trabajo a elección. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para ser válido.
Reemplaza YOUR_BUCKET_NAME por 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

Cuando ejecutes esta plantilla, necesitarás la ruta de Cloud Storage a la plantilla:

gs://dataflow-templates/latest/Word_Count

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

En este ejemplo, debes reemplazar los siguientes valores:

Reemplaza YOUR_PROJECT_ID por el ID del proyecto.
Reemplaza JOB_NAME por un nombre de trabajo a elección. El nombre del trabajo debe coincidir con la expresión regular [a-z]([-a-z0-9]{0,38}[a-z0-9])? para ser válido.
Reemplaza YOUR_BUCKET_NAME por 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" }
}

Código fuente de la plantilla

Java

Ver en GitHub Comentarios

/*
 * Copyright (C) 2016 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.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();
  }
}