Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.
Guida rapida: invia flussi di messaggi da Pub/Sub utilizzando Dataflow

Trasmetti flussi di messaggi da Pub/Sub utilizzando Dataflow

Dataflow è un servizio completamente gestito per la trasformazione e l'arricchimento dei dati in modalità flusso (in tempo reale) e batch con pari affidabilità ed espressività. Offre un ambiente di sviluppo della pipeline semplificato grazie all'SDK Apache Beam, con un ricco set di primitive di windowing e analisi delle sessioni nonché un ecosistema di connettori di origine e sink. Questa guida rapida mostra come utilizzare Dataflow per:

  • Leggere i messaggi pubblicati in un argomento Pub/Sub
  • Raggruppa (o raggruppa) i messaggi in base al timestamp
  • Scrivi i messaggi su Cloud Storage

Questa guida rapida spiega come utilizzare Dataflow in Java e Python. È supportato anche SQL. Questa guida rapida è disponibile anche come tutorial su Google Cloud Skills Boost che offre credenziali temporanee per iniziare.

Puoi anche iniziare a utilizzare i modelli di Dataflow basati sull'interfaccia utente se non intendi eseguire il trattamento personalizzato dei dati.

Prima di iniziare

  1. Accedi al tuo account Google Cloud. Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti gratuiti per l'esecuzione, il test e il deployment dei carichi di lavoro.
  2. Installa Google Cloud CLI.

  3. Per inizializzare l'interfaccia a riga di comando gcloud, esegui il comando seguente: 

    gcloud init

  4. Crea o seleziona un progetto Google Cloud.

    • Creare un progetto Cloud:

      gcloud projects create PROJECT_ID

    • Seleziona il progetto Cloud che hai creato:

      gcloud config set project PROJECT_ID

  5. Assicurati che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata su un progetto.

  6. Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON API, Pub/Sub, Resource Manager, and Cloud Scheduler. 

    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

  7. Configurare l'autenticazione:

    1. Crea l'account di servizio: 

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      Sostituisci SERVICE_ACCOUNT_NAME con un nome per l'account di servizio.

    2. Concedi i ruoli all'account di servizio. Esegui il comando seguente una volta per ciascuno dei seguenti ruoli IAM: 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

      Sostituisci quanto segue:

      • SERVICE_ACCOUNT_NAME: nome dell'account di servizio
      • PROJECT_ID: l'ID progetto in cui hai creato l'account di servizio
      • ROLE: il ruolo da concedere

    3. Concedi al tuo Account Google un ruolo che ti consenta di utilizzare i ruoli dell'account di servizio e collegarlo ad altre risorse: 

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

      Sostituisci quanto segue:

      • SERVICE_ACCOUNT_NAME: nome dell'account di servizio
      • PROJECT_ID: l'ID progetto in cui hai creato l'account di servizio
      • USER_EMAIL: l'indirizzo email del tuo Account Google
    8.
  8. Installa Google Cloud CLI.

  9. Per inizializzare l'interfaccia a riga di comando gcloud, esegui il comando seguente: 

    gcloud init

  10. Crea o seleziona un progetto Google Cloud.

    • Creare un progetto Cloud:

      gcloud projects create PROJECT_ID

    • Seleziona il progetto Cloud che hai creato:

      gcloud config set project PROJECT_ID

  11. Assicurati che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata su un progetto.

  12. Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON API, Pub/Sub, Resource Manager, and Cloud Scheduler. 

    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

  13. Configurare l'autenticazione:

    1. Crea l'account di servizio: 

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      Sostituisci SERVICE_ACCOUNT_NAME con un nome per l'account di servizio.

    2. Concedi i ruoli all'account di servizio. Esegui il comando seguente una volta per ciascuno dei seguenti ruoli IAM: 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

      Sostituisci quanto segue:

      • SERVICE_ACCOUNT_NAME: nome dell'account di servizio
      • PROJECT_ID: l'ID progetto in cui hai creato l'account di servizio
      • ROLE: il ruolo da concedere

    3. Concedi al tuo Account Google un ruolo che ti consenta di utilizzare i ruoli dell'account di servizio e collegarlo ad altre risorse: 

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

      Sostituisci quanto segue:

      • SERVICE_ACCOUNT_NAME: nome dell'account di servizio
      • PROJECT_ID: l'ID progetto in cui hai creato l'account di servizio
      • USER_EMAIL: l'indirizzo email del tuo Account Google
    14.

  14. Crea le credenziali di autenticazione per il tuo Account Google: 

    gcloud auth application-default login

Configurare il progetto Pub/Sub

  1. Crea variabili per il bucket, il progetto e la regione. I nomi dei bucket Cloud Storage devono essere univoci a livello globale. Seleziona una regione di Dataflow vicino alla posizione in cui esegui i comandi in questa guida rapida. Il valore della variabile REGION deve essere un nome di una regione valido. Per saperne di più sulle regioni e sulle località, consulta Località di 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 bucket Cloud Storage appartenente a questo progetto: 

    gsutil mb gs://$BUCKET_NAME

  3. Crea un argomento Pub/Sub in questo progetto: 

    gcloud pubsub topics create $TOPIC_ID

  4. Creare un job Cloud Scheduler in questo progetto. Il job pubblica un messaggio in un argomento Pub/Sub a intervalli di 1 minuto.

    Se non esiste un'app App Engine per il progetto, questo passaggio ne creerà uno. 

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

    Avvia il job. 

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

  5. Utilizza i comandi seguenti per clonare il repository della guida rapida e vai alla directory del codice di esempio:

    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

Trasmetti flussi di messaggi da Pub/Sub a Cloud Storage

Esempio di codice

Questo codice campione utilizza Dataflow per:

  • Leggere i messaggi Pub/Sub.
  • Finestra (o gruppo) di messaggi in intervalli di dimensioni fisse pubblicando timestamp.

  • Scrivi i messaggi in ogni finestra in file in 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.PipelineOptions;
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 PipelineOptions, 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("utf-8"))

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,
    )

Avvia la pipeline

Per avviare la pipeline, esegui il comando seguente:

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

Il comando precedente viene eseguito localmente e avvia un job Dataflow che viene eseguito nel cloud. Quando il comando restituisce JOB_MESSAGE_DETAILED: Workers have started successfully, esci dal programma locale utilizzando Ctrl+C.

Osserva l'avanzamento di job e pipeline

Puoi osservare l'avanzamento del job nella console di Dataflow.

Vai alla console Dataflow

Osserva i progressi del job

Apri la visualizzazione dei dettagli del job per vedere:

  • Struttura del job
  • Log job
  • Metriche della fase

Osserva i progressi del job

Potresti dover attendere qualche minuto per vedere i file di output in Cloud Storage.

Osserva i progressi del job

In alternativa, utilizza la riga di comando riportata di seguito per controllare quali file sono stati scritti.

gsutil ls gs://${BUCKET_NAME}/samples/

L'output dovrebbe avere l'aspetto seguente:

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

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questa pagina, elimina il progetto Cloud con le risorse.

  1. Elimina il job Cloud Scheduler.

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

  2. Nella console di Dataflow, arresta il job. Annulla la pipeline senza svuotarla.

  3. Eliminare l'argomento.

    gcloud pubsub topics delete $TOPIC_ID

  4. Elimina i file creati dalla pipeline.

    gsutil -m rm -rf "gs://${BUCKET_NAME}/samples/output*"
gsutil -m rm -rf "gs://${BUCKET_NAME}/temp/*"

  5. Rimuovi il bucket Cloud Storage.

    gsutil rb gs://${BUCKET_NAME}

  6. Elimina l'account di servizio: 
    gcloud iam service-accounts delete SERVICE_ACCOUNT_EMAIL

  7. (Facoltativo) Revoca le credenziali di autenticazione che hai creato ed elimina il file delle credenziali locale. 

    gcloud auth application-default revoke

  8. Facoltativo: revoca le credenziali dall'interfaccia a riga di comando gcloud. 

    gcloud auth revoke

Passaggi successivi