Transmitir mensajes de Pub/Sub mediante Dataflow y Cloud Storage

Transmitir mensajes de Pub/Sub mediante Dataflow y Cloud Storage

Dataflow es un servicio totalmente gestionado que está diseñado para transformar y enriquecer datos tanto en streaming (en tiempo real) como por lotes con la misma fiabilidad y expresividad. Proporciona un entorno de desarrollo de canalizaciones simplificado que usa el SDK de Apache Beam, que tiene un amplio conjunto de tipos primitivos de análisis de ventanas y sesiones, así como un ecosistema de conectores de origen y de receptor. En esta guía de inicio rápido se muestra cómo usar Dataflow para hacer lo siguiente:

• Leer mensajes publicados en un tema de Pub/Sub
• Ventana (o agrupa) los mensajes por marca de tiempo.
• Escribir los mensajes en Cloud Storage

En esta guía de inicio rápido se explica cómo usar Dataflow en Java y Python. También se admite SQL. Esta guía de inicio rápido también se ofrece como un tutorial de Acelerador de conocimientos de Google Cloud, que proporciona credenciales temporales para que puedas empezar.

También puedes empezar usando plantillas de Dataflow basadas en la interfaz de usuario si no tienes intención de hacer un tratamiento de datos personalizado.

Nota: Antes de empezar esta guía de inicio rápido, asegúrate de que estás usando una versión de Python o una versión de Java que sea compatible con Dataflow.

Antes de empezar

Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
Install the Google Cloud CLI.
Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.
Para inicializar gcloud CLI, ejecuta el siguiente comando:
gcloud init
Create or select a Google Cloud project.
Roles required to select or create a project
• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.
• Create a Google Cloud project:

  gcloud projects create PROJECT_ID

  Replace PROJECT_ID with a name for the Google Cloud project you are creating.

• Select the Google Cloud project that you created:

  gcloud config set project PROJECT_ID

  Replace PROJECT_ID with your Google Cloud project name.
Verify that billing is enabled for your Google Cloud project.
Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON API, Pub/Sub, Resource Manager, and Cloud Scheduler APIs:
Roles required to enable APIs
To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.
gcloud services enable dataflow.googleapis.com  compute.googleapis.com  logging.googleapis.com  storage-component.googleapis.com  storage-api.googleapis.com  pubsub.googleapis.com  cloudresourcemanager.googleapis.com  cloudscheduler.googleapis.com
Set up authentication:
1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator). Learn how to grant roles.

2. Create the service account:

   gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

   Replace SERVICE_ACCOUNT_NAME with a name for the service account.

3. Grant roles to the service account. Run the following command once for each of the following IAM roles: roles/dataflow.worker, roles/storage.objectAdmin, roles/pubsub.admin:

   gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=ROLE

   Replace the following:

   • SERVICE_ACCOUNT_NAME: the name of the service account
   • PROJECT_ID: the project ID where you created the service account
   • ROLE: the role to grant
   Note: The --role flag affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later.

4. Grant the required role to the principal that will attach the service account to other resources.

   gcloud iam service-accounts add-iam-policy-binding SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com --member="user:USER_EMAIL" --role=roles/iam.serviceAccountUser

   Replace the following:

   • SERVICE_ACCOUNT_NAME: the name of the service account
   • PROJECT_ID: the project ID where you created the service account
   • USER_EMAIL: the email address for a Google Account

Install the Google Cloud CLI.

Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.

Para inicializar gcloud CLI, ejecuta el siguiente comando:

gcloud init

Create or select a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

• Create a Google Cloud project:

  gcloud projects create PROJECT_ID

  Replace PROJECT_ID with a name for the Google Cloud project you are creating.

• Select the Google Cloud project that you created:

  gcloud config set project PROJECT_ID

  Replace PROJECT_ID with your Google Cloud project name.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON API, Pub/Sub, Resource Manager, and Cloud Scheduler APIs:

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

gcloud services enable dataflow.googleapis.com  compute.googleapis.com  logging.googleapis.com  storage-component.googleapis.com  storage-api.googleapis.com  pubsub.googleapis.com  cloudresourcemanager.googleapis.com  cloudscheduler.googleapis.com

Set up authentication:

1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator). Learn how to grant roles.

2. Create the service account:

   gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

   Replace SERVICE_ACCOUNT_NAME with a name for the service account.

3. Grant roles to the service account. Run the following command once for each of the following IAM roles: roles/dataflow.worker, roles/storage.objectAdmin, roles/pubsub.admin:

   gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=ROLE

   Replace the following:

   • SERVICE_ACCOUNT_NAME: the name of the service account
   • PROJECT_ID: the project ID where you created the service account
   • ROLE: the role to grant
   Note: The --role flag affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later.

4. Grant the required role to the principal that will attach the service account to other resources.

   gcloud iam service-accounts add-iam-policy-binding SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com --member="user:USER_EMAIL" --role=roles/iam.serviceAccountUser

   Replace the following:

   • SERVICE_ACCOUNT_NAME: the name of the service account
   • PROJECT_ID: the project ID where you created the service account
   • USER_EMAIL: the email address for a Google Account

1. Create local authentication credentials for your user account:

   gcloud auth application-default login

   If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

Configurar el proyecto de Pub/Sub

1. Crea variables para tu segmento, proyecto y región. Los nombres de los segmentos de Cloud Storage deben ser únicos de forma global. Selecciona una región de Dataflow cercana a la ubicación en la que ejecutas los comandos de esta guía de inicio rápido. El valor de la variable REGION debe ser un nombre de región válido. Para obtener más información sobre las regiones y las ubicaciones, consulta el artículo sobre las ubicaciones de Dataflow.

   BUCKET_NAME=BUCKET_NAME
   PROJECT_ID=$(gcloud config get-value project)
   TOPIC_ID=TOPIC_ID
   REGION=DATAFLOW_REGION
   SERVICE_ACCOUNT=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

2. Crea un segmento de Cloud Storage propiedad de este proyecto:

   gcloud storage buckets create gs://$BUCKET_NAME

3. Crea un tema de Pub/Sub en este proyecto:

   gcloud pubsub topics create $TOPIC_ID

4. Crea una tarea de Cloud Scheduler en este proyecto. La tarea publica un mensaje en un tema de Pub/Sub a intervalos de un minuto.

   Si no existe ninguna aplicación de App Engine para el proyecto, en este paso se creará una.

   gcloud scheduler jobs create pubsub publisher-job --schedule="* * * * *" \
       --topic=$TOPIC_ID --message-body="Hello!" --location=$REGION

   Inicia el trabajo.

   gcloud scheduler jobs run publisher-job --location=$REGION

5. Usa los siguientes comandos para clonar el repositorio de inicio rápido y ve al directorio del código de ejemplo:

   Java

   git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
   cd java-docs-samples/pubsub/streaming-analytics

   Python

   git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
   cd python-docs-samples/pubsub/streaming-analytics
   pip install -r requirements.txt  # Install Apache Beam dependencies

Enviar mensajes de Pub/Sub a Cloud Storage

Código de ejemplo

Este código de ejemplo usa Dataflow para hacer lo siguiente:

• Leer mensajes de Pub/Sub.
• Ventana (o agrupa) los mensajes en intervalos de tamaño fijo por marcas de tiempo de publicación.

• Escribe los mensajes de cada ventana en archivos de Cloud Storage.

Java

import java.io.IOException;
import org.apache.beam.examples.common.WriteOneFilePerWindow;
import org.apache.beam.sdk.Pipeline;
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.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

public class PubSubToGcs {
  /*
   * Define your own configuration options. Add your own arguments to be processed
   * by the command-line parser, and specify default values for them.
   */
  public interface PubSubToGcsOptions extends StreamingOptions {
    @Description("The Cloud Pub/Sub topic to read from.")
    @Required
    String getInputTopic();

    void setInputTopic(String value);

    @Description("Output file's window size in number of minutes.")
    @Default.Integer(1)
    Integer getWindowSize();

    void setWindowSize(Integer value);

    @Description("Path of the output file including its filename prefix.")
    @Required
    String getOutput();

    void setOutput(String value);
  }

  public static void main(String[] args) throws IOException {
    // The maximum number of shards when writing output.
    int numShards = 1;

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

    options.setStreaming(true);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        // 1) Read string messages from a Pub/Sub topic.
        .apply("Read PubSub Messages", PubsubIO.readStrings().fromTopic(options.getInputTopic()))
        // 2) Group the messages into fixed-sized minute intervals.
        .apply(Window.into(FixedWindows.of(Duration.standardMinutes(options.getWindowSize()))))
        // 3) Write one file to GCS for every window of messages.
        .apply("Write Files to GCS", new WriteOneFilePerWindow(options.getOutput(), numShards));

    // Execute the pipeline and wait until it finishes running.
    pipeline.run().waitUntilFinish();
  }
}

Python

import argparse
from datetime import datetime
import logging
import random

from apache_beam import (
    DoFn,
    GroupByKey,
    io,
    ParDo,
    Pipeline,
    PTransform,
    WindowInto,
    WithKeys,
)
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows


class GroupMessagesByFixedWindows(PTransform):
    """A composite transform that groups Pub/Sub messages based on publish time
    and outputs a list of tuples, each containing a message and its publish time.
    """

    def __init__(self, window_size, num_shards=5):
        # Set window size to 60 seconds.
        self.window_size = int(window_size * 60)
        self.num_shards = num_shards

    def expand(self, pcoll):
        return (
            pcoll
            # Bind window info to each element using element timestamp (or publish time).
            | "Window into fixed intervals"
            >> WindowInto(FixedWindows(self.window_size))
            | "Add timestamp to windowed elements" >> ParDo(AddTimestamp())
            # Assign a random key to each windowed element based on the number of shards.
            | "Add key" >> WithKeys(lambda _: random.randint(0, self.num_shards - 1))
            # Group windowed elements by key. All the elements in the same window must fit
            # memory for this. If not, you need to use `beam.util.BatchElements`.
            | "Group by key" >> GroupByKey()
        )


class AddTimestamp(DoFn):
    def process(self, element, publish_time=DoFn.TimestampParam):
        """Processes each windowed element by extracting the message body and its
        publish time into a tuple.
        """
        yield (
            element.decode("utf-8"),
            datetime.utcfromtimestamp(float(publish_time)).strftime(
                "%Y-%m-%d %H:%M:%S.%f"
            ),
        )


class WriteToGCS(DoFn):
    def __init__(self, output_path):
        self.output_path = output_path

    def process(self, key_value, window=DoFn.WindowParam):
        """Write messages in a batch to Google Cloud Storage."""

        ts_format = "%H:%M"
        window_start = window.start.to_utc_datetime().strftime(ts_format)
        window_end = window.end.to_utc_datetime().strftime(ts_format)
        shard_id, batch = key_value
        filename = "-".join([self.output_path, window_start, window_end, str(shard_id)])

        with io.gcsio.GcsIO().open(filename=filename, mode="w") as f:
            for message_body, publish_time in batch:
                f.write(f"{message_body},{publish_time}\n".encode())


def run(input_topic, output_path, window_size=1.0, num_shards=5, pipeline_args=None):
    # Set `save_main_session` to True so DoFns can access globally imported modules.
    pipeline_options = PipelineOptions(
        pipeline_args, streaming=True, save_main_session=True
    )

    with Pipeline(options=pipeline_options) as pipeline:
        (
            pipeline
            # Because `timestamp_attribute` is unspecified in `ReadFromPubSub`, Beam
            # binds the publish time returned by the Pub/Sub server for each message
            # to the element's timestamp parameter, accessible via `DoFn.TimestampParam`.
            # https://beam.apache.org/releases/pydoc/current/apache_beam.io.gcp.pubsub.html#apache_beam.io.gcp.pubsub.ReadFromPubSub
            | "Read from Pub/Sub" >> io.ReadFromPubSub(topic=input_topic)
            | "Window into" >> GroupMessagesByFixedWindows(window_size, num_shards)
            | "Write to GCS" >> ParDo(WriteToGCS(output_path))
        )


if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input_topic",
        help="The Cloud Pub/Sub topic to read from."
        '"projects/<PROJECT_ID>/topics/<TOPIC_ID>".',
    )
    parser.add_argument(
        "--window_size",
        type=float,
        default=1.0,
        help="Output file's window size in minutes.",
    )
    parser.add_argument(
        "--output_path",
        help="Path of the output GCS file including the prefix.",
    )
    parser.add_argument(
        "--num_shards",
        type=int,
        default=5,
        help="Number of shards to use when writing windowed elements to GCS.",
    )
    known_args, pipeline_args = parser.parse_known_args()

    run(
        known_args.input_topic,
        known_args.output_path,
        known_args.window_size,
        known_args.num_shards,
        pipeline_args,
    )

Iniciar el flujo de procesamiento

Para iniciar la canalización, ejecuta el siguiente comando:

Java

mvn compile exec:java \
  -Dexec.mainClass=com.examples.pubsub.streaming.PubSubToGcs \
  -Dexec.cleanupDaemonThreads=false \
  -Dexec.args=" \
    --project=$PROJECT_ID \
    --region=$REGION \
    --inputTopic=projects/$PROJECT_ID/topics/$TOPIC_ID \
    --output=gs://$BUCKET_NAME/samples/output \
    --gcpTempLocation=gs://$BUCKET_NAME/temp \
    --runner=DataflowRunner \
    --windowSize=2 \
    --serviceAccount=$SERVICE_ACCOUNT"

Python

python PubSubToGCS.py \
  --project=$PROJECT_ID \
  --region=$REGION \
  --input_topic=projects/$PROJECT_ID/topics/$TOPIC_ID \
  --output_path=gs://$BUCKET_NAME/samples/output \
  --runner=DataflowRunner \
  --window_size=2 \
  --num_shards=2 \
  --temp_location=gs://$BUCKET_NAME/temp \
  --service_account_email=$SERVICE_ACCOUNT

El comando anterior se ejecuta de forma local e inicia una tarea de Dataflow que se ejecuta en la nube. Cuando el comando devuelva JOB_MESSAGE_DETAILED: Workers have started successfully, sal del programa local con Ctrl+C.

Observar el progreso de las tareas y las canalizaciones

Puedes observar el progreso de la tarea en la consola de Dataflow.

Ir a la consola de Dataflow

Abre la vista de detalles del trabajo para ver lo siguiente:

• Estructura de la tarea
• Registros de tareas
• Métricas de etapa

Es posible que tengas que esperar unos minutos para ver los archivos de salida en Cloud Storage.

También puedes usar la línea de comandos que aparece a continuación para comprobar qué archivos se han escrito.

gcloud storage ls gs://${BUCKET_NAME}/samples/

La salida debería tener este aspecto:

Java

gs://{$BUCKET_NAME}/samples/output-22:30-22:32-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:32-22:34-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:34-22:36-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:36-22:38-0-of-1

Python

gs://{$BUCKET_NAME}/samples/output-22:30-22:32-0
gs://{$BUCKET_NAME}/samples/output-22:30-22:32-1
gs://{$BUCKET_NAME}/samples/output-22:32-22:34-0
gs://{$BUCKET_NAME}/samples/output-22:32-22:34-1

Limpieza

Para evitar que se apliquen cargos en tu Google Cloud cuenta por los recursos utilizados en esta página, elimina el Google Cloud proyecto con los recursos.

1. Elimina la tarea de Cloud Scheduler.

   gcloud scheduler jobs delete publisher-job --location=$REGION

2. En la consola de Dataflow, detén el trabajo. Cancela la canalización sin vaciarla.

3. Elimina el tema.

   gcloud pubsub topics delete $TOPIC_ID

4. Elimina los archivos creados por la canalización.

   gcloud storage rm "gs://${BUCKET_NAME}/samples/output*" --recursive --continue-on-error
   gcloud storage rm "gs://${BUCKET_NAME}/temp/*" --recursive --continue-on-error

5. Elimina el segmento de Cloud Storage.

   gcloud storage rm gs://${BUCKET_NAME} --recursive

6. Elimina la cuenta de servicio:

   gcloud iam service-accounts delete SERVICE_ACCOUNT_EMAIL

7. Optional: Revoke the authentication credentials that you created, and delete the local credential file.

   gcloud auth application-default revoke

8. Optional: Revoke credentials from the gcloud CLI.

   gcloud auth revoke

Siguientes pasos

• Si quieres visualizar los mensajes de Pub/Sub en función de una marca de tiempo personalizada, puedes especificarla como atributo en el mensaje de Pub/Sub y, a continuación, usarla con withTimestampAttribute de PubsubIO.

• Consulta las plantillas de Dataflow de código abierto de Google diseñadas para el streaming.

• Consulta más información sobre cómo se integra Dataflow con Pub/Sub.

• Consulta este tutorial para leer datos de Pub/Sub y escribirlos en BigQuery con plantillas flexibles de Dataflow.

• Para obtener más información sobre las ventanas, consulta el ejemplo de pipeline de juegos para móviles de Apache Beam.