Aus Apache Kafka in Dataflow lesen

In diesem Dokument wird beschrieben, wie Sie Daten aus Apache Kafka in Dataflow lesen. Außerdem finden Sie hier Tipps zur Leistungsoptimierung und Best Practices.

In den meisten Anwendungsfällen empfiehlt es sich, den Managed I/O-Connector zu verwenden, um Daten aus Kafka zu lesen.

Wenn Sie eine erweiterte Leistungsoptimierung benötigen, können Sie den KafkaIO Connector verwenden. Der KafkaIO Connector ist für Java oder über das mehrsprachige Pipeline-Framework für Python und Go verfügbar.

Parallelität

In den folgenden Abschnitten wird beschrieben, wie Sie die Parallelität beim Lesen aus Kafka konfigurieren.

Übersicht

Die Parallelität wird durch zwei Faktoren begrenzt: die maximale Anzahl von Workern (max_num_workers) und die Anzahl der Kafka-Partitionen. In Dataflow ist standardmäßig ein Parallelitäts-Fanout von 4 × max_num_workers festgelegt. Der Fanout ist jedoch durch die Anzahl der Partitionen begrenzt. Wenn beispielsweise 100 vCPUs verfügbar sind, die Pipeline aber nur aus 10 Kafka-Partitionen liest, beträgt die maximale Parallelität 10.

Um die Parallelität zu maximieren, sollten Sie mindestens 4 × max_num_workers Kafka-Partitionen haben. Wenn Ihr Job den Portable Runner verwendet, können Sie die Parallelität noch weiter erhöhen. Ein guter Ausgangspunkt ist, die Anzahl der Partitionen auf das Doppelte der Anzahl der Worker-vCPUs festzulegen.

Umverteilen

Wenn Sie die Anzahl der Partitionen nicht erhöhen können, können Sie die Parallelität erhöhen, indem Sie aufrufen KafkaIO.Read.withRedistribute. Mit dieser Methode wird der Pipeline eine Redistribute-Transformation hinzugefügt, die Dataflow einen Hinweis gibt, die Daten effizienter neu zu verteilen und zu parallelisieren. Wir empfehlen dringend, die optimale Anzahl von Shards anzugeben, indem Sie aufrufen KafkaIO.Read.withRedistributeNumKeys. Wenn Sie nur KafkaIO.Read.withRedistribute verwenden, können zahlreiche Schlüssel generiert werden, was zu Leistungsproblemen führen kann. Weitere Informationen finden Sie unter Phasen mit hoher Parallelität identifizieren. Durch das Umverteilen der Daten entsteht zusätzlicher Overhead für die Durchführung des Shuffle-Schritts. Weitere Informationen finden Sie unter Zusammenführung verhindern.

Um die Kosten des Umverteilungs-Shuffles zu minimieren, rufen Sie KafkaIO.Read.withOffsetDeduplication auf. In diesem Modus wird die Menge der Daten minimiert, die im Rahmen des Shuffles beibehalten werden müssen, während gleichzeitig eine genau einmalige Verarbeitung erfolgt.

Wenn eine genau einmalige Verarbeitung nicht erforderlich ist, können Sie Duplikate zulassen, indem Sie KafkaIO.Read.withAllowDuplicates aufrufen.

In der folgenden Tabelle sind die Optionen für die Umverteilung zusammengefasst:

Lastverteilung

Achten Sie darauf, dass die Last zwischen den Partitionen relativ gleichmäßig verteilt ist. Wenn die Last ungleichmäßig verteilt ist, kann dies zu einer schlechten Auslastung der Worker führen. Worker, die aus Partitionen mit geringerer Last lesen, sind möglicherweise relativ inaktiv, während Worker, die aus Partitionen mit hoher Last lesen, in Verzug geraten können. Dataflow bietet Messwerte für den Rückstand pro Partition.

Wenn die Last ungleichmäßig verteilt ist, kann der dynamische Arbeitsausgleich helfen, die Arbeit zu verteilen. Dataflow kann beispielsweise einen Worker zuweisen, der aus mehreren Partitionen mit geringem Volumen liest, und einen anderen Worker zuweisen, der aus einer einzelnen Partition mit hohem Volumen liest. Zwei Worker können jedoch nicht aus derselben Partition lesen. Eine stark ausgelastete Partition kann daher dazu führen, dass sich die Verarbeitung der Pipeline verzögert.

Best Practices

Dieser Abschnitt enthält Empfehlungen zum Lesen von Daten aus Kafka in Dataflow.

Themen mit geringem Volumen

Ein häufiges Szenario ist das gleichzeitige Lesen aus vielen Themen mit geringem Volumen, z. B. ein Thema pro Kunde. Das Erstellen separater Dataflow-Jobs für jedes Thema ist kostspielig, da für jeden Job mindestens ein vollständiger Worker erforderlich ist. Verwenden Sie stattdessen eine der folgenden Optionen:

• Themen zusammenführen. Themen kombinieren, bevor sie in Dataflow aufgenommen werden. Das Aufnehmen einiger Themen mit hohem Volumen ist viel effizienter als das Aufnehmen vieler Themen mit geringem Volumen. Jedes Thema mit hohem Volumen kann von einem einzelnen Dataflow-Job verarbeitet werden, der seine Worker vollständig nutzt.

• Mehrere Themen lesen. Wenn Sie Themen nicht kombinieren können, bevor Sie sie in Dataflow aufnehmen, können Sie eine Pipeline erstellen, die aus mehreren Themen liest. Mit diesem Ansatz kann Dataflow mehreren Themen denselben Worker zuweisen. Es gibt zwei Möglichkeiten, diesen Ansatz zu implementieren:

  • Ein Schritt zum Lesen. Erstellen Sie eine einzelne Instanz des KafkaIO-Connectors und konfigurieren Sie ihn so, dass er mehrere Themen liest. Filtern Sie dann nach dem Namen des Themas, um für jedes Thema eine andere Logik anzuwenden. Beispielcode finden Sie unter Aus mehreren Themen lesen. Verwenden Sie diese Option, wenn sich alle Themen im selben Cluster befinden. Ein Nachteil ist, dass Probleme mit einer einzelnen Datensenke oder einer einzelnen Transformation dazu führen können, dass sich bei allen Themen ein Rückstand ansammelt.

    Für komplexere Anwendungsfälle übergeben Sie eine Reihe von KafkaSourceDescriptor-Objekten, die die Themen angeben, aus denen gelesen werden soll. Mit KafkaSourceDescriptor können Sie die Themenliste bei Bedarf später aktualisieren. Für diese Funktion ist Java mit dem Portable Runner erforderlich.

  • Mehrere Schritte zum Lesen. Wenn Sie aus Themen lesen möchten, die sich in verschiedenen Clustern befinden, kann Ihre Pipeline mehrere KafkaIO-Instanzen enthalten. Während der Job ausgeführt wird, können Sie einzelne Quellen mithilfe von Transformationszuordnungenaktualisieren. Das Festlegen eines neuen Themas oder Clusters wird nur unterstützt, wenn Sie den Portable Runner verwenden. Die Beobachtbarkeit kann bei diesem Ansatz eine Herausforderung sein, da Sie jede einzelne Lesetransformation überwachen müssen, anstatt sich auf Messwerte auf Pipeline-Ebene zu verlassen.

Zurück zu Kafka committen

Standardmäßig verwendet der KafkaIO-Connector keine Kafka-Offsets, um den Fortschritt zu verfolgen, und committet nicht zurück zu Kafka. Wenn Sie commitOffsetsInFinalize aufrufen, versucht der Connector, nach dem Committen von Datensätzen in Dataflow zurück zu Kafka zu committen. In Dataflow committete Datensätze werden möglicherweise nicht vollständig verarbeitet. Wenn Sie die Pipeline abbrechen, wird möglicherweise ein Offset committet, ohne dass die Datensätze jemals vollständig verarbeitet wurden.

Da mit enable.auto.commit=True Offsets committet werden, sobald sie aus Kafka gelesen werden, ohne dass sie von Dataflow verarbeitet werden, wird die Verwendung dieser Option nicht empfohlen. Wir empfehlen, sowohl enable.auto.commit=False als auch commitOffsetsInFinalize=True festzulegen. Wenn Sie enable.auto.commit auf True setzen, können Daten verloren gehen, wenn die Pipeline während der Verarbeitung unterbrochen wird. Bereits in Kafka committete Datensätze werden möglicherweise gelöscht.

Wasserzeichen

Standardmäßig verwendet der KafkaIO Connector die aktuelle Verarbeitungszeit, um das Ausgabewasserzeichen und die Ereigniszeit zuzuweisen. Wenn Sie dieses Verhalten ändern möchten, rufen Sie withTimestampPolicyFactory auf und weisen Sie eine TimestampPolicy zu. Beam bietet Implementierungen von TimestampPolicy, die das Wasserzeichen entweder anhand der Protokollanhängezeit von Kafka oder der Erstellungszeit der Nachricht berechnen.

Überlegungen zum Runner

Der KafkaIO Connector hat zwei zugrunde liegende Implementierungen für Kafka-Lesevorgänge: die ältere ReadFromKafkaViaUnbounded und die neuere ReadFromKafkaViaSDF. Dataflow wählt automatisch die beste Implementierung für Ihren Job basierend auf der SDK-Sprache und den Jobanforderungen aus. Fordern Sie nicht explizit einen Runner oder eine Kafka-Implementierung an, es sei denn, Sie benötigen bestimmte Funktionen, die nur in dieser Implementierung verfügbar sind. Weitere Informationen zur Auswahl eines Runners finden Sie unter Dataflow Portable Runner verwenden.

Wenn Ihre Pipeline withTopic oder withTopics, fragt die ältere Implementierung Kafka zum Zeitpunkt der Pipeline-Erstellung nach den verfügbaren Partitionen ab. Die Maschine, die die Pipeline erstellt, muss die Berechtigung haben, eine Verbindung zu Kafka herzustellen. Wenn Sie einen Berechtigungsfehler erhalten, prüfen Sie, ob Sie die Berechtigung haben, lokal eine Verbindung zu Kafka herzustellen. Sie können dieses Problem vermeiden, indem Sie withTopicPartitions verwenden, das zum Zeitpunkt der Pipeline-Erstellung keine Verbindung zu Kafka herstellt.

Für Produktion bereitstellen

Wenn Sie Ihre Lösung in der Produktion bereitstellen, empfiehlt es sich, flexible Vorlagen zu verwenden. Wenn Sie eine flexible Vorlage verwenden, wird die Pipeline in einer einheitlichen Umgebung gestartet. Dies kann dazu beitragen, lokale Konfigurationsprobleme zu vermeiden.

Das Logging von KafkaIO kann sehr ausführlich sein. Sie können die Loggingebene in der Produktion so reduzieren:

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

Weitere Informationen finden Sie unter Logebenen für Pipeline-Worker festlegen.

Netzwerk konfigurieren

Normalerweise startet Dataflow Instanzen in Ihrem VPC-Standardnetzwerk (Virtual Private Cloud). Je nach Kafka-Konfiguration müssen Sie möglicherweise ein anderes Netzwerk und Subnetz für Dataflow konfigurieren. Weitere Informationen finden Sie unter Netzwerk und Subnetzwerk angeben. Erstellen Sie beim Konfigurieren Ihres Netzwerks Firewallregeln, mit denen die Dataflow-Worker-Maschinen die Kafka-Broker erreichen können.

Wenn Sie VPC Service Controls verwenden, platzieren Sie den Kafka-Cluster innerhalb des VPC Service Controls-Perimeters oder weiten Sie die Perimeter auf das autorisierte VPN oder Cloud Interconnect aus.

Wenn Ihr Kafka-Cluster außerhalb von Google Cloudbereitgestellt wird, müssen Sie eine Netzwerkverbindung zwischen Dataflow und dem Kafka-Cluster erstellen. Es gibt mehrere Netzwerkoptionen mit unterschiedlichen Vor- und Nachteilen:

• Verbindung über einen freigegebenen RFC 1918-Adressbereich herstellen, indem Sie eine der folgenden Optionen verwenden:
  • Dedicated Interconnect
  • IPsec VPN (virtuelles privates Netzwerk)
• Sie können Ihren extern gehosteten Kafka-Cluster über öffentliche IP-Adressen erreichen, indem Sie eine der folgenden Optionen verwenden:
  • Öffentliches Internet
  • Direct Peering
  • Carrier Peering

Dedicated Interconnect ist die beste Option für eine vorhersagbare Leistung und Zuverlässigkeit. Die Einrichtung kann jedoch länger dauern, da die neuen Verbindungen durch Dritte bereitgestellt werden müssen. Mit einer Topologie auf Basis einer öffentlichen IP-Adresse können Sie schnell starten, da nur wenig Netzwerkarbeit erforderlich ist.

In den nächsten beiden Abschnitten werden diese Optionen ausführlicher beschrieben.

Freigegebener RFC 1918-Adressraum

Sowohl Dedicated Interconnect als auch IPsec-VPN bieten Ihnen direkten Zugriff auf RFC 1918-IP-Adressen in Ihrer Virtual Private Cloud (VPC), was Ihre Kafka-Konfiguration vereinfachen kann. Wenn Sie eine VPN-basierte Topologie verwenden, sollten Sie ein VPN mit hohem Durchsatz erstellen.

Normalerweise startet Dataflow Instanzen in Ihrem standardmäßigen VPC-Netzwerk. In einer privaten Netzwerktopologie mit explizit in Cloud Router definierten Routen , die Subnetzwerke in Google Cloud mit diesem Kafka-Cluster verbinden, müssen Sie die Standorte Ihrer Dataflow-Instanzen besser steuern können. Sie können mit Dataflow die network und subnetwork Ausführungsparameter konfigurieren.

Achten Sie darauf, dass das entsprechende Subnetzwerk über genügend IP-Adressen verfügt, damit Dataflow Instanzen beim horizontalen Skalieren starten kann. Wenn Sie zum Starten der Dataflow-Instanzen ein separates Netzwerk erstellen, ist außerdem wichtig, dass Sie eine Firewallregel haben, die TCP-Traffic zwischen allen virtuellen Maschinen im Projekt zulässt. Im Standardnetzwerk ist diese Firewallregel bereits konfiguriert.

Öffentlicher IP-Adressraum

Diese Architektur verwendet zum Schutz des Traffics zwischen externen Clients und Kafka Transport Layer Security (TLS). Die Kommunikation zwischen den Brokern erfolgt in Klartext. Wenn sich der Kafka-Listener an eine Netzwerkschnittstelle bindet, die sowohl für die interne als auch für die externe Kommunikation verwendet wird, ist die Konfiguration des Listeners unkompliziert. In vielen Szenarien unterscheiden sich die extern veröffentlichten Adressen der Kafka-Broker im Cluster jedoch von den internen Netzwerkschnittstellen, die Kafka verwendet. In solchen Szenarien können Sie das Attribut advertised.listeners verwenden:

# 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

Externe Clients verbinden sich über Port 9093 über einen „SSL“-Kanal her und interne Clients verbinden sich über einen Klartextkanal über Port 9092. Wenn Sie unter advertised.listeners eine Adresse angeben, verwenden Sie DNS-Namen (in diesem Beispiel kafkabroker-n.mydomain.com), die für externen und internen Traffic in dieselbe Instanz aufgelöst werden. Öffentliche IP-Adressen funktionieren eventuell nicht, da die Adressen möglicherweise für den internen Traffic nicht richtig aufgelöst werden.

Kafka optimieren

Die Einstellungen für Ihren Kafka-Cluster und Kafka-Client können einen großen Einfluss auf die Leistung haben. Insbesondere die folgenden Einstellungen sind möglicherweise zu niedrig. Dieser Abschnitt enthält einige Vorschläge für die ersten Schritte. Sie sollten jedoch mit diesen Werten für Ihre spezielle Arbeitslast experimentieren.

• unboundedReaderMaxElements. Die Standardeinstellung ist 10.000. Ein höherer Wert wie 100.000 kann die Größe der Bundleserhöhen,was die Leistung erheblich verbessern kann, wenn Ihre Pipeline Aggregationen enthält. Ein höherer Wert kann jedoch auch die Latenz erhöhen. Verwenden Sie setUnboundedReaderMaxElements, um den Wert festzulegen. Diese Einstellung gilt nicht für den Portable Runner. Verwenden Sie für den Portable Runner die sdf_checkpoint_after_output_bytes Dataflow-Dienstoption.

• unboundedReaderMaxReadTimeMs. Die Standardeinstellung ist 10.000 ms. Ein höherer Wert wie 20.000 ms kann die Bundle-Größe erhöhen, während ein niedrigerer Wert wie 5.000 ms die Latenz oder den Rückstand reduzieren kann. Verwenden Sie setUnboundedReaderMaxReadTimeMs, um den Wert festzulegen. Diese Einstellung gilt nicht für den Portable Runner. Verwenden Sie für den Portable Runner die sdf_checkpoint_after_duration Dataflow-Dienstoption.

• max.poll.records. Die Standardeinstellung ist 500. Ein höherer Wert kann zu einer besseren Leistung führen, da mehr eingehende Datensätze gleichzeitig abgerufen werden, insbesondere wenn Sie den Portable Runner verwenden. Rufen Sie withConsumerConfigUpdates auf, um den Wert festzulegen.

• fetch.max.bytes. Die Standardeinstellung ist 1 MB. Ein höherer Wert kann den Durchsatz verbessern, da die Anzahl der Anfragen reduziert wird, insbesondere wenn Sie den Portable Runner verwenden. Wenn Sie den Wert jedoch zu hoch festlegen, kann die Latenz steigen. Die nachgelagerte Verarbeitung ist jedoch eher der Hauptengpass. Ein empfohlener Startwert ist 100 MB. Rufen Sie withConsumerConfigUpdates auf, um den Wert festzulegen.

• max.partition.fetch.bytes. Die Standardeinstellung ist 1 MB. Mit diesem Parameter wird die maximale Datenmenge pro Partition festgelegt, die der Server zurückgibt. Wenn Sie den Wert erhöhen, kann der Durchsatz verbessert werden, da die Anzahl der Anfragen reduziert wird, insbesondere wenn Sie den Portable Runner verwenden. Wenn Sie den Wert jedoch zu hoch festlegen, kann die Latenz steigen. Die nachgelagerte Verarbeitung ist jedoch eher der Hauptengpass. Ein empfohlener Startwert ist 100 MB. Rufen Sie withConsumerConfigUpdates auf, um den Wert festzulegen.

• consumerPollingTimeout. Die Standardeinstellung ist 2 Sekunden. Wenn die Zeitüberschreitung des Verbraucherclients eintritt, bevor Datensätze gelesen werden können, versuchen Sie, einen höheren Wert festzulegen. Diese Einstellung ist am häufigsten relevant, wenn Sie regionsübergreifende Lesevorgänge oder Lesevorgänge mit einem langsamen Netzwerk ausführen. Rufen Sie withConsumerPollingTimeout auf, um den Wert festzulegen.

Achten Sie darauf, dass receive.buffer.bytes groß genug ist, um die Größe der Nachrichten zu verarbeiten. Wenn der Wert zu klein ist, wird in den Logs möglicherweise angezeigt, dass Verbraucher ständig neu erstellt werden und einen bestimmten Offset suchen.

Beispiele

Die folgenden Codebeispiele zeigen, wie Sie Dataflow-Pipelines erstellen, die Daten aus Kafka lesen. Wenn Sie die Standardanmeldedaten für Anwendungen in Verbindung mit dem vom Google Cloud Managed Service for Apache Kafka bereitgestellten Callback-Handler verwenden, ist die kafka-clients Version 3.7.0 oder höher erforderlich.

Aus einem einzelnen Thema lesen

In diesem Beispiel wird der Managed I/O-Connector verwendet. Es wird gezeigt, wie Sie Daten aus einem Kafka-Thema lesen und die Nachrichten-Nutzlasten in Textdateien schreiben.

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

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

Aus mehreren Themen lesen

In diesem Beispiel wird der KafkaIO-Connector verwendet. Es wird gezeigt, wie Sie Daten aus mehreren Kafka-Themen lesen und für jedes Thema eine separate Pipeline-Logik anwenden.

Für komplexere Anwendungsfälle können Sie dynamisch eine Reihe von KafkaSourceDescriptor-Objekten übergeben, damit Sie die Liste der Themen, aus denen gelesen werden soll, aktualisieren können. Für diesen Ansatz ist Java mit dem Portable Runner erforderlich.

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

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

Nächste Schritte

• Leistungsmerkmale von Apache Kafka-zu-BigQuery-Pipelines
• In Apache Kafka schreiben.
• Weitere Informationen zu Managed I/O