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:

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 con 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

Selecciona the WordCount template en el menú desplegable Plantilla de Dataflow.
Ingresa un nombre para el trabajo en el campo Nombre del trabajo.
Ingresa los valores de tus parámetros en los campos de parámetros provistos.
Haz clic en Ejecutar trabajo.

gcloud

Ejecuta con la herramienta de línea de comandos de gcloud.

Nota: Si quieres usar la herramienta de línea de comandos de gcloud para ejecutar plantillas, debes tener la versión 138.0.0 o superior del SDK de Cloud.

Cuando ejecutas esta plantilla, necesitas la ruta de acceso de Cloud Storage a la plantilla:

gs://dataflow-templates/latest/Word_Count

Ejecuta el siguiente comando:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Word_Count \
    --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
BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.

API

Ejecuta con la API de REST.

Cuando ejecutas esta plantilla, necesitas la ruta de acceso 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.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/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 de tu proyecto
JOB_NAME: Es el nombre del trabajo que elijas
BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.

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();
  }
}