Transmita mensagens do Pub/Sub através do Dataflow e do Cloud Storage

Transmita mensagens do Pub/Sub através do Dataflow e do Cloud Storage

O Dataflow é um serviço totalmente gerido para transformar e enriquecer dados nos modos de streaming (em tempo real) e em lote com igual fiabilidade e expressividade. Oferece um ambiente de desenvolvimento de pipeline simplificado através do SDK do Apache Beam, que tem um conjunto avançado de primitivas de análise de janelas e sessões, bem como um ecossistema de conetores de origem e destino. Este início rápido mostra como usar o Dataflow para:

• Ler mensagens publicadas num tópico do Pub/Sub
• Agrupar as mensagens por indicação de tempo
• Escreva as mensagens no Cloud Storage

Este início rápido apresenta a utilização do Dataflow em Java e Python. O SQL também é suportado. Este início rápido também é oferecido como um tutorial do Google Cloud Skills Boost que oferece credenciais temporárias para começar.

Também pode começar por usar modelos do Dataflow baseados na IU se não pretender fazer o tratamento de dados personalizado.

Nota: antes de iniciar este início rápido, certifique-se de que está a usar uma versão do Python ou uma versão do Java compatível com o Dataflow.

Antes de começar

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.
Se estiver a usar um fornecedor de identidade (IdP) externo, tem primeiro de iniciar sessão na CLI gcloud com a sua identidade federada.
Para inicializar a CLI gcloud, execute o seguinte 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 role (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) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). 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.

Se estiver a usar um fornecedor de identidade (IdP) externo, tem primeiro de iniciar sessão na CLI gcloud com a sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte 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 role (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) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). 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.

Configure o seu projeto Pub/Sub

1. Crie variáveis para o seu contentor, projeto e região. Os nomes dos contentores do Cloud Storage têm de ser globalmente exclusivos. Selecione uma região do Dataflow perto de onde executa os comandos neste início rápido. O valor da variável REGION tem de ser um nome de região válido. Para mais informações sobre regiões e localizações, consulte o artigo Localizações do 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. Crie um contentor do Cloud Storage pertencente a este projeto:

   gcloud storage buckets create gs://$BUCKET_NAME

3. Crie um tópico Pub/Sub neste projeto:

   gcloud pubsub topics create $TOPIC_ID

4. Crie um trabalho do Cloud Scheduler neste projeto. A tarefa publica uma mensagem num tópico Pub/Sub em intervalos de 1 minuto.

   Se não existir uma app do App Engine para o projeto, este passo cria uma.

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

   Inicie a tarefa.

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

5. Use os seguintes comandos para clonar o repositório de início rápido e navegar para o diretório de código de exemplo:

   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

Transmita mensagens do Pub/Sub para o Cloud Storage

Exemplo de código

Este exemplo de código usa o Dataflow para:

• Ler mensagens do Pub/Sub.
• Divida as mensagens em intervalos de tamanho fixo por indicações de tempo de publicação.

• Escrever as mensagens em cada janela para ficheiros no 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,
    )

Inicie a conduta

Para iniciar o pipeline, execute o seguinte 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

O comando anterior é executado localmente e inicia uma tarefa do Dataflow que é executada na nuvem. Quando o comando devolver JOB_MESSAGE_DETAILED: Workers have started successfully, saia do programa local com Ctrl+C.

Observe o progresso do trabalho e do pipeline

Pode observar o progresso da tarefa na consola do Dataflow.

Aceda à consola do Dataflow

Abra a vista de detalhes do trabalho para ver:

• Estrutura do trabalho
• Registos de tarefas
• Métricas de palco

Pode ter de aguardar alguns minutos para ver os ficheiros de saída no Cloud Storage.

Em alternativa, use a linha de comando abaixo para verificar que ficheiros foram escritos.

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

O resultado deve ter o seguinte aspeto:

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

Limpar

Para evitar incorrer em custos na sua Google Cloud conta pelos recursos usados nesta página, elimine o Google Cloud projeto com os recursos.

1. Elimine a tarefa do Cloud Scheduler.

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

2. Na consola do Dataflow, pare a tarefa. Cancele o pipeline sem o esvaziar.

3. Elimine o tópico.

   gcloud pubsub topics delete $TOPIC_ID

4. Elimine os ficheiros criados pelo pipeline.

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

5. Remova o contentor do Cloud Storage.

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

6. Elimine a conta de serviço:

   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

O que se segue?

• Se quiser colocar as mensagens do Pub/Sub numa janela por uma data/hora personalizada, pode especificar a data/hora como um atributo na mensagem do Pub/Sub e, em seguida, usar a data/hora personalizada com o do PubsubIO.withTimestampAttribute

• Consulte os modelos de código aberto do Dataflow da Google concebidos para streaming.

• Leia mais sobre como o Dataflow se integra com o Pub/Sub.

• Consulte este tutorial que lê a partir do Pub/Sub e escreve no BigQuery através de modelos flexíveis do Dataflow.

• Para saber mais sobre a criação de janelas, consulte o exemplo Apache Beam Mobile Gaming Pipeline.