Leggere da Apache Kafka a Dataflow

Questo documento descrive come leggere i dati da Apache Kafka a Dataflow e include suggerimenti sulle prestazioni e best practice.

Per la maggior parte dei casi d'uso, valuta la possibilità di utilizzare il connettore I/O gestito per leggere da Kafka.

Se hai bisogno di una messa a punto delle prestazioni più avanzata, valuta la possibilità di utilizzare il KafkaIO connettore. Il connettore KafkaIO è disponibile per Java o utilizzando il framework delle pipeline multilingue per Python e Go.

Parallelismo

Le sezioni seguenti descrivono come configurare il parallelismo durante la lettura da Kafka.

Panoramica

Il parallelismo è vincolato da due fattori: il numero massimo di worker (max_num_workers) e il numero di partizioni Kafka. Per impostazione predefinita, Dataflow utilizza un fan-out di parallelismo di 4 x max_num_workers. Tuttavia, il fan-out è limitato dal numero di partizioni. Ad esempio, se sono disponibili 100 vCPU, ma la pipeline legge solo da 10 partizioni Kafka, il parallelismo massimo è 10.

Per massimizzare il parallelismo, ti consigliamo di avere almeno 4 x max_num_workers partizioni Kafka. Se il job utilizza il runner portatile, valuta la possibilità di impostare un parallelismo ancora più elevato. Un buon punto di partenza è avere partizioni pari al doppio del numero di vCPU worker.

Ridistribuisci

Se non puoi aumentare il numero di partizioni, puoi aumentare il parallelismo chiamando KafkaIO.Read.withRedistribute. Questo metodo aggiunge una trasformazione Redistribute alla pipeline, che fornisce un suggerimento a Dataflow per ridistribuire e parallelizzare i dati in modo più efficiente. Ti consigliamo vivamente di specificare il numero ottimale di shard chiamando KafkaIO.Read.withRedistributeNumKeys. L'utilizzo di KafkaIO.Read.withRedistribute da solo può generare numerose chiavi, causando problemi di prestazioni. Per ulteriori informazioni, consulta Identificare le fasi con parallelismo elevato. La ridistribuzione dei dati aggiunge un overhead aggiuntivo per eseguire il passaggio di shuffle. Per ulteriori informazioni, consulta Impedire la fusione.

Per ridurre al minimo il costo dello shuffle di ridistribuzione, chiama KafkaIO.Read.withOffsetDeduplication. Questa modalità riduce al minimo la quantità di dati da rendere persistenti nell'ambito dello shuffle, pur fornendo un'elaborazione exactly-once.

Se l'elaborazione exactly-once non è obbligatoria, puoi consentire i duplicati chiamando KafkaIO.Read.withAllowDuplicates.

La tabella seguente riassume le opzioni di ridistribuzione:

Opzione Modalità di elaborazione Apache Beam Configurazione
Ridistribuisci input Exactly-once v2.60+ KafkaIO.Read.withRedistribute()
Consenti duplicati At-least-once v2.60+ KafkaIO.Read.withRedistribute().withAllowDuplicates()
Deduplicazione dell'offset Exactly-once v2.69+ KafkaIO.Read.withRedistribute().withOffsetDeduplication()

Skew del carico

Cerca di assicurarti che il carico tra le partizioni sia relativamente uniforme e non distorto. Se il carico è distorto, può comportare un utilizzo insufficiente dei worker. I worker che leggono da partizioni con un carico più leggero potrebbero essere relativamente inattivi, mentre i worker che leggono da partizioni con un carico elevato potrebbero rimanere indietro. Dataflow fornisce metriche per il backlog per partizione.

Se il carico è distorto, il bilanciamento dinamico del lavoro può aiutarti a distribuire il lavoro. Ad esempio, Dataflow potrebbe allocare un worker per leggere da più partizioni a basso volume e allocare un altro worker per leggere da una singola partizione a volume elevato. Tuttavia, due worker non possono leggere dalla stessa partizione, quindi una partizione con un carico elevato può comunque causare il ritardo della pipeline.

Best practice

Questa sezione contiene consigli per la lettura da Kafka in Dataflow.

Argomenti a basso volume

Uno scenario comune è leggere da molti argomenti a basso volume contemporaneamente, ad esempio un argomento per cliente. La creazione di job Dataflow separati per ogni argomento è inefficiente in termini di costi, perché ogni job richiede almeno un worker completo. In alternativa, valuta le seguenti opzioni:

  • Unisci gli argomenti. Combina gli argomenti prima di importarli in Dataflow. L'importazione di alcuni argomenti a volume elevato è molto più efficiente dell'importazione di molti argomenti a basso volume. Ogni argomento a volume elevato può essere gestito da un singolo job Dataflow che utilizza completamente i suoi worker.

  • Leggi più argomenti. Se non puoi combinare gli argomenti prima di importarli in Dataflow, valuta la possibilità di creare una pipeline che legge da più argomenti. Questo approccio consente a Dataflow di assegnare più argomenti allo stesso worker. Esistono due modi per implementare questo approccio:

    • Singolo passaggio di lettura. Crea una singola istanza del connettore KafkaIO e configurala per leggere più argomenti. Poi filtra per nome dell'argomento per applicare una logica diversa per ogni argomento. Per un esempio di codice, consulta Leggere da più argomenti. Valuta questa opzione se tutti gli argomenti si trovano nello stesso cluster. Uno svantaggio è che i problemi con un singolo sink o una singola trasformazione potrebbero causare l'accumulo di backlog in tutti gli argomenti.

      Per casi d'uso più avanzati, passa un insieme di oggetti KafkaSourceDescriptor che specificano gli argomenti da cui leggere. L'utilizzo di KafkaSourceDescriptor ti consente di aggiornare l'elenco degli argomenti in un secondo momento, se necessario. Questa funzionalità richiede Java con il runner portatile.

    • Più passaggi di lettura. Per leggere da argomenti che si trovano in cluster diversi, la pipeline può includere più istanze KafkaIO. Durante l'esecuzione del job, puoi aggiornare le singole origini utilizzando i mapping di trasformazione. L'impostazione di un nuovo argomento o cluster è supportata solo quando si utilizza il runner portatile. L'osservabilità è una potenziale sfida con questo approccio, perché devi monitorare ogni singola trasformazione di lettura anziché affidarti alle metriche a livello di pipeline.

Eseguire il commit di nuovo in Kafka

Per impostazione predefinita, il connettore KafkaIO non utilizza gli offset Kafka per monitorare l'avanzamento e non esegue il commit di nuovo in Kafka. Se chiami commitOffsetsInFinalize, il connettore fa del suo meglio per eseguire il commit di nuovo in Kafka dopo che i record sono stati sottoposti a commit in Dataflow. I record sottoposti a commit in Dataflow potrebbero non essere elaborati completamente, quindi se annulli la pipeline, è possibile che venga eseguito il commit di un offset senza che i record vengano mai elaborati completamente.

Poiché l'impostazione di enable.auto.commit=True esegue il commit degli offset non appena vengono letti da Kafka senza alcuna elaborazione da parte di Dataflow, l'utilizzo di questa opzione non è consigliato. Ti consigliamo di impostare sia enable.auto.commit=False sia commitOffsetsInFinalize=True. Se imposti enable.auto.commit su True, i dati possono essere persi se la pipeline viene interrotta durante l'elaborazione. I record già sottoposti a commit su Kafka potrebbero essere eliminati.

Filigrane

Per impostazione predefinita, il connettore KafkaIO utilizza l'ora di elaborazione corrente per assegnare la filigrana di output e l'ora dell'evento. Per modificare questo comportamento, chiama withTimestampPolicyFactory e assegna un TimestampPolicy. Beam fornisce implementazioni di TimestampPolicy che calcolano la filigrana in base all'ora di aggiunta dei log di Kafka o all'ora di creazione del messaggio.

Considerazioni sul runner

Il connettore KafkaIO ha due implementazioni sottostanti per le letture Kafka: la versione precedente ReadFromKafkaViaUnbounded e la versione più recente ReadFromKafkaViaSDF. Dataflow sceglie automaticamente l'implementazione migliore per il tuo job in base alla lingua dell'SDK e ai requisiti del job. Evita di richiedere esplicitamente un runner o un'implementazione Kafka, a meno che non ti servano funzionalità specifiche disponibili solo in quell'implementazione. Per ulteriori informazioni sulla scelta di un runner, consulta Utilizzare il runner portatile Dataflow.

Se la pipeline utilizza withTopic o withTopics, l'implementazione precedente esegue una query su Kafka al momento della creazione della pipeline per le partizioni disponibili. La macchina che crea la pipeline deve avere l'autorizzazione per connettersi a Kafka. Se ricevi un errore di autorizzazione, verifica di avere le autorizzazioni per connetterti a Kafka localmente. Puoi evitare questo problema utilizzando withTopicPartitions, che non si connette a Kafka al momento della creazione della pipeline.

Distribuzione in produzione

Quando esegui il deployment della soluzione in produzione, ti consigliamo di utilizzare i modelli Flex. Utilizzando un modello flessibile, la pipeline viene avviata da un ambiente coerente, il che può contribuire a mitigare i problemi di configurazione locale.

La registrazione di KafkaIO può essere piuttosto dettagliata. Valuta la possibilità di ridurre il livello di logging in produzione nel seguente modo:

sdkHarnessLogLevelOverrides='{"org.apache.kafka.clients.consumer.internals.SubscriptionState":"WARN"}'.

Per ulteriori informazioni, consulta Impostare i livelli di log dei worker della pipeline.

Configura il networking

Per impostazione predefinita, Dataflow avvia le istanze nella rete Virtual Private Cloud (VPC) predefinita. A seconda della configurazione di Kafka, potrebbe essere necessario configurare una rete e una subnet diverse per Dataflow. Per ulteriori informazioni, consulta Specificare una rete e una subnet. Quando configuri la rete, crea regole firewall che consentano alle macchine worker di Dataflow di raggiungere i broker Kafka.

Se utilizzi Controlli di servizio VPC, inserisci il cluster Kafka all'interno del perimetro dei Controlli di servizio VPC oppure estendi i perimetri alla VPN o a Cloud Interconnect autorizzati.

Se il cluster Kafka è sottoposto a deployment al di fuori di Google Cloud, devi creare una connessione di rete tra Dataflow e il cluster Kafka. Esistono diverse opzioni di networking con compromessi diversi:

Dedicated Interconnect è l'opzione migliore per prestazioni e affidabilità prevedibili, ma la configurazione può richiedere più tempo perché le terze parti devono eseguire il provisioning dei nuovi circuiti. Con una topologia basata su indirizzi IP pubblici, puoi iniziare rapidamente perché non è necessario eseguire molte operazioni di networking.

Le due sezioni successive descrivono queste opzioni in modo più dettagliato.

Spazio di indirizzi RFC 1918 condiviso

Sia Dedicated Interconnect sia IPsec VPN ti consentono di accedere direttamente agli indirizzi IP RFC 1918 nel tuo Virtual Private Cloud (VPC), il che può semplificare la configurazione di Kafka. Se utilizzi una topologia basata su VPN, valuta la possibilità di configurare una VPN a velocità effettiva elevata.

Per impostazione predefinita, Dataflow avvia le istanze nella rete VPC predefinita. In una topologia di rete privata con route definite esplicitamente in Cloud Router che collegano le subnet in Google Cloud a quel cluster Kafka, hai bisogno di un maggiore controllo sulla posizione delle istanze Dataflow. Puoi utilizzare Dataflow per configurare i network e subnetwork parametri di esecuzione.

Assicurati che la subnet corrispondente abbia indirizzi IP sufficienti per consentire a Dataflow di avviare le istanze quando tenta di eseguire lo scale out. Inoltre, quando crei una rete separata per avviare le istanze Dataflow, assicurati di avere una regola firewall che abiliti il traffico TCP tra tutte le macchine virtuali nel progetto. La rete predefinita ha già questa regola firewall configurata.

Spazio di indirizzi IP pubblici

Questa architettura utilizza Transport Layer Security (TLS) per proteggere il traffico tra i client esterni e Kafka e utilizza il traffico non criptato per la comunicazione tra i broker. Quando il listener Kafka si associa a un'interfaccia di rete utilizzata sia per la comunicazione interna sia per quella esterna, la configurazione del listener è semplice. Tuttavia, in molti scenari, gli indirizzi pubblicizzati esternamente dei broker Kafka nel cluster differiscono dalle interfacce di rete interne utilizzate da Kafka. In questi scenari, puoi utilizzare la proprietà advertised.listeners:

# Configure protocol map
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:SSL

# Use plaintext for inter-broker communication
inter.broker.listener.name=INTERNAL

# Specify that Kafka listeners should bind to all local interfaces
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093

# Separately, specify externally visible address
advertised.listeners=INTERNAL://kafkabroker-n.mydomain.com:9092,EXTERNAL://kafkabroker-n.mydomain.com:9093

I client esterni si connettono utilizzando la porta 9093 tramite un canale "SSL" e i client interni si connettono utilizzando la porta 9092 tramite un canale di testo normale. Quando specifichi un indirizzo in advertised.listeners, utilizza nomi DNS (kafkabroker-n.mydomain.com, in questo esempio) che si risolvono nella stessa istanza sia per il traffico esterno sia per quello interno. L'utilizzo di indirizzi IP pubblici potrebbe non funzionare perché gli indirizzi potrebbero non essere risolti per il traffico interno.

Ottimizzare Kafka

Le impostazioni del cluster Kafka e del client Kafka possono avere un impatto significativo sulle prestazioni. In particolare, le seguenti impostazioni potrebbero essere troppo basse. Questa sezione fornisce alcuni punti di partenza suggeriti, ma ti consigliamo di sperimentare questi valori per il tuo workload specifico.

  • unboundedReaderMaxElements. Il valore predefinito è 10.000. Un valore più alto, ad esempio 100.000, può aumentare le dimensioni dei bundle, il che può migliorare notevolmente le prestazioni se la pipeline include aggregazioni. Tuttavia, un valore più alto potrebbe anche aumentare la latenza. Per impostare il valore, utilizza setUnboundedReaderMaxElements. Questa impostazione non si applica al runner portatile. Per il runner portatile, utilizza l' sdf_checkpoint_after_output_bytes opzione del servizio Dataflow.

  • unboundedReaderMaxReadTimeMs. Il valore predefinito è 10.000 ms. Un valore più alto, ad esempio 20.000 ms, può aumentare le dimensioni del bundle, mentre un valore più basso, ad esempio 5000 ms, può ridurre la latenza o il backlog. Per impostare il valore, utilizza setUnboundedReaderMaxReadTimeMs. Questa impostazione non si applica al runner portatile. Per il runner portatile, utilizza l'opzione del servizio Dataflow sdf_checkpoint_after_duration.

  • max.poll.records. Il valore predefinito è 500. Un valore più alto potrebbe avere un rendimento migliore recuperando più record in entrata contemporaneamente, soprattutto quando si utilizza il runner portatile. Per impostare il valore, chiama withConsumerConfigUpdates.

  • fetch.max.bytes. Il valore predefinito è 1 MB. Un valore più alto potrebbe migliorare la velocità effettiva riducendo il numero di richieste, soprattutto quando si utilizza il runner portatile. Tuttavia, se lo imposti su un valore troppo alto, la latenza potrebbe aumentare, anche se è più probabile che l'elaborazione downstream sia il collo di bottiglia principale. Un valore iniziale consigliato è 100 MB. Per impostare il valore, chiama withConsumerConfigUpdates.

  • max.partition.fetch.bytes. Il valore predefinito è 1 MB. Questo parametro imposta la quantità massima di dati per partizione restituita dal server. L'aumento del valore può migliorare la velocità effettiva riducendo il numero di richieste, soprattutto quando si utilizza il runner portatile. Tuttavia, se lo imposti su un valore troppo alto, la latenza potrebbe aumentare, anche se è più probabile che l'elaborazione downstream sia il collo di bottiglia principale. Un valore iniziale consigliato è 100 MB. Per impostare il valore, chiama withConsumerConfigUpdates.

  • consumerPollingTimeout. Il valore predefinito è 2 secondi. Se il client consumer va in timeout prima di poter leggere i record, prova a impostare un valore più alto. Questa impostazione è più spesso pertinente quando si eseguono letture tra regioni o letture con una rete lenta. Per impostare il valore, chiama withConsumerPollingTimeout.

Assicurati che receive.buffer.bytes sia sufficientemente grande da gestire le dimensioni dei messaggi. Se il valore è troppo piccolo, i log potrebbero mostrare che i consumer vengono ricreati continuamente e cercano un offset specifico.

Esempi

I seguenti esempi di codice mostrano come creare pipeline Dataflow che leggono da Kafka. Quando utilizzi le credenziali predefinite dell'applicazione insieme al gestore di callback fornito da Google Cloud Managed Service per Apache Kafka, kafka-clients è richiesta la versione 3.7.0 o successive.

Leggere da un singolo argomento

Questo esempio utilizza il connettore I/O gestito. Mostra come leggere da un argomento Kafka e scrivere i payload dei messaggi nei file di testo.

Java

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import com.google.common.collect.ImmutableMap;
import java.io.UnsupportedEncodingException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.managed.Managed;
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.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class KafkaRead {

  public static Pipeline createPipeline(Options options) {

    // Create configuration parameters for the Managed I/O transform.
    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("bootstrap_servers", options.getBootstrapServer())
        .put("topic", options.getTopic())
        .put("format", "RAW")
        .put("max_read_time_seconds", 15)
        .put("auto_offset_reset_config", "earliest")
        .build();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    pipeline
        // Read messages from Kafka.
        .apply(Managed.read(Managed.KAFKA).withConfig(config)).getSinglePCollection()
        // Get the payload of each message and convert to a string.
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via((row -> {
              var bytes = row.getBytes("payload");
              try {
                return new String(bytes, "UTF-8");
              } catch (UnsupportedEncodingException e) {
                throw new RuntimeException(e);
              }
            })))
        // Write the payload to a text file.
        .apply(TextIO
            .write()
            .to(options.getOutputPath())
            .withSuffix(".txt")
            .withNumShards(1));
    return pipeline;
  }
}

Python

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import argparse

import apache_beam as beam

from apache_beam import window
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #     --topic=$KAFKA_TOPIC --bootstrap_server=$BOOTSTRAP_SERVER
    #     --output=$CLOUD_STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument("--topic")
            parser.add_argument("--bootstrap_server")
            parser.add_argument("--output")

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        (
            pipeline
            # Read messages from an Apache Kafka topic.
            | beam.managed.Read(
                beam.managed.KAFKA,
                config={
                  "bootstrap_servers": options.bootstrap_server,
                  "topic": options.topic,
                  "data_format": "RAW",
                  "auto_offset_reset_config": "earliest",
                  # The max_read_time_seconds parameter is intended for testing.
                  # Avoid using this parameter in production.
                  "max_read_time_seconds": 5
                }
            )
            # Subdivide the output into fixed 5-second windows.
            | beam.WindowInto(window.FixedWindows(5))
            | WriteToText(
                file_path_prefix=options.output, file_name_suffix=".txt", num_shards=1
            )
        )

Leggere da più argomenti

Questo esempio utilizza il connettore KafkaIO. Mostra come leggere da più argomenti Kafka e applicare una logica di pipeline separata per ogni argomento.

Per casi d'uso più avanzati, passa dinamicamente un insieme di KafkaSourceDescriptor oggetti, in modo da poter aggiornare l'elenco degli argomenti da cui leggere. Questo approccio richiede Java con il runner portatile.

Java

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.kafka.KafkaIO;
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.transforms.Filter;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.joda.time.Duration;
import org.joda.time.Instant;

public class KafkaReadTopics {

  public static Pipeline createPipeline(Options options) {
    String topic1 = options.getTopic1();
    String topic2 = options.getTopic2();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    var allTopics = pipeline
        .apply(KafkaIO.<Long, String>read()
            .withTopics(List.of(topic1, topic2))
            .withBootstrapServers(options.getBootstrapServer())
            .withKeyDeserializer(LongDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
            .withMaxReadTime(Duration.standardSeconds(10))
            .withStartReadTime(Instant.EPOCH)
        );

    // Create separate pipeline branches for each topic.
    // The first branch filters on topic1.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic1)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic1)
            .withSuffix(".txt")
            .withNumShards(1)
        );

    // The second branch filters on topic2.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic2)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic2)
            .withSuffix(".txt")
            .withNumShards(1)
        );
    return pipeline;
  }
}

Python

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import argparse

import apache_beam as beam

from apache_beam.io.kafka import ReadFromKafka
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #   --bootstrap_server=$BOOTSTRAP_SERVER --output=$STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument('--bootstrap_server')
            parser.add_argument('--output')

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        # Read from two Kafka topics.
        all_topics = pipeline | ReadFromKafka(consumer_config={
                "bootstrap.servers": options.bootstrap_server
            },
            topics=["topic1", "topic2"],
            with_metadata=True,
            max_num_records=10,
            start_read_time=0
        )

        # Filter messages from one topic into one branch of the pipeline.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic1')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic1" >> WriteToText(
                file_path_prefix=options.output + '/topic1/output',
                file_name_suffix='.txt',
                num_shards=1))

        # Filter messages from the other topic.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic2')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic2" >> WriteToText(
                file_path_prefix=options.output + '/topic2/output',
                file_name_suffix='.txt',
                num_shards=1))

Passaggi successivi