View a markdown version of this page

Verwenden von Kafka-Streams mit MSK Serverless - Amazon Managed Streaming für Apache Kafka
Eine Kafka Streams-Anwendung erstellen

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Verwenden von Kafka-Streams mit MSK Serverless

Anmerkung

Wenn Sie MSK Standard-Broker oder Express-Broker verwenden, können Sie Kafka Streams verwenden, ohne interne Themen manuell zu erstellen. Die Anleitung auf dieser Seite bezieht sich nur auf MSK Serverless.

Kafka Streams unterstützt zustandslose und statusbehaftete Transformationen. Zustandsorientierte Transformationen wie Count, Aggregate oder Join verwenden Operatoren, die ihren Status in internen Kafka-Themen speichern. Darüber hinaus speichern einige zustandslose Transformationen wie GroupBy oder Repartition ihre Ergebnisse in internen Kafka-Themen. Standardmäßig benennt Kafka Streams diese internen Themen auf der Grundlage des entsprechenden Operators. Wenn diese Themen nicht existieren, erstellt Kafka Streams interne Kafka-Themen. Für die Erstellung der internen Themen codiert Kafka Streams die segment.bytes-Konfiguration fest und legt sie auf 50 MB fest. MSK Serverless schützt einige Themenkonfigurationen, einschließlich segment.size, während der Themenerstellung. Daher kann eine Kafka Streams-Anwendung mit statusbehafteten Transformationen die internen Themen nicht mithilfe von MSK Serverless erstellen.

Um solche Kafka Streams-Anwendungen auf MSK Serverless auszuführen, müssen Sie die internen Themen selbst erstellen. Identifizieren und benennen Sie dazu zunächst die Kafka Streams-Operatoren, für die Themen erforderlich sind. Erstellen Sie dann die entsprechenden internen Kafka-Themen.

Anmerkung

  • Es hat sich bewährt, die Operatoren in Kafka Streams manuell zu benennen, insbesondere diejenigen, die von internen Themen abhängen. Informationen zur Benennung von Operatoren finden Sie unter Benennen von Operatoren in einer Kafka Streams-DSL-Anwendung in der Kafka Streams-Dokumentation.

  • Der interne Themenname für eine Stateful-Transformation hängt von der Version der Kafka Streams-Anwendung und dem Namen application.id des Stateful-Operators ab. application.id-statefuloperator_name

Eine Kafka Streams-Anwendung mit MSK Serverless erstellen

Wenn Ihre Kafka Streams-Anwendung auf application.id eingestellt istmsk-streams-processing, können Sie mit MSK Serverless eine Kafka Streams-Anwendung erstellen. Verwenden Sie dazu den count() Operator, der ein internes Thema mit dem Namen erfordert. Beispiel, msk-streams-processing-count-store.

Gehen Sie wie folgt vor, um eine Kafka Streams-Anwendung zu erstellen:

Identifizieren und benennen Sie die Operatoren

  1. Identifizieren Sie die Stateful-Prozessoren anhand der Stateful-Transformationen in der Kafka Streams-Dokumentation.

    Einige Beispiele für Stateful-Prozessoren sind,, oder. count aggregate join

  2. Identifizieren Sie die Prozessoren, die Themen für die Neupartitionierung erstellen.

    Das folgende Beispiel enthält eine count() Operation, für die ein Status erforderlich ist.

    var stream =
    paragraphStream
        .groupByKey()
        .count()
        .toStream();

  3. Um dem Thema einen Namen zu geben, fügen Sie einen Namen für jeden statusbehafteten Prozessor hinzu. Je nach Prozessortyp erfolgt die Benennung durch eine andere Benennungsklasse. Eine count() Operation ist beispielsweise eine Aggregationsoperation. Daher benötigt es die Materialized Klasse.

    Informationen zu den Benennungsklassen für die statusbehafteten Operationen finden Sie unter Fazit in der Dokumentation zu Kafka Streams.

    Im folgenden Beispiel wird der Name des count() Operators so festgelegt, dass er die count-store Materialized Klasse verwendet.

    var stream =
    paragraphStream
        .groupByKey()
        .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count-store")  // descriptive name for the store
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()))
        .toStream();

Erstellen Sie die internen Themen

Kafka Streams-Präfixe application.id für Namen interner Themen, wobei dies benutzerdefiniert application.id ist. Beispiel, application.id-internal_topic_name. Die internen Themen sind normale Kafka-Themen, und Sie können die Themen mithilfe der Informationen erstellen, die in Erstellen Sie ein Apache Kafka-Thema oder AdminClient über die Kafka-API verfügbar sind.

Je nach Anwendungsfall können Sie die standardmäßigen Bereinigungs- und Aufbewahrungsrichtlinien von Kafka Streams verwenden oder deren Werte anpassen. Sie definieren diese in und. cleanup.policy retention.ms

Im folgenden Beispiel werden die Themen mit der AdminClient API erstellt und der Wert application.id auf gesetztmsk-streams-processing.

try (AdminClient client = AdminClient.create(configs.kafkaProps())) {
    Collection<NewTopic> topics = new HashSet<>();
    topics.add(new NewTopic("msk-streams-processing-count-store", 3, (short) 3));
    client.createTopics(topics);
}

Nachdem die Themen auf dem Cluster erstellt wurden, kann Ihre Kafka Streams-Anwendung das msk-streams-processing-count-store Thema für den count() Vorgang verwenden.

(Optional) Überprüfen Sie den Namen des Themas

Sie können den Topologie-Deskriptor verwenden, um die Topologie Ihres Streams zu beschreiben und die Namen der internen Themen einzusehen. Das folgende Beispiel zeigt, wie der Topologie-Deskriptor ausgeführt wird.

final StreamsBuilder builder = new StreamsBuilder();
Topology topology = builder.build();
System.out.println(topology.describe());

Die folgende Ausgabe zeigt die Topologie des Streams für das vorherige Beispiel.

Topology Description:
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input_topic])
      --> KSTREAM-AGGREGATE-0000000001
    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])
      --> KTABLE-TOSTREAM-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KTABLE-TOSTREAM-0000000002 (stores: [])
      --> KSTREAM-SINK-0000000003
      <-- KSTREAM-AGGREGATE-0000000001
    Sink: KSTREAM-SINK-0000000003 (topic: output_topic)
      <-- KTABLE-TOSTREAM-0000000002

Informationen zur Verwendung des Topologiebeschreibers finden Sie unter Benennen von Operatoren in einer Kafka Streams-DSL-Anwendung in der Kafka Streams-Dokumentation.

Beispiele für Benennungsoperatoren

Dieser Abschnitt enthält einige Beispiele für Benennungsoperatoren.

Beispiel für einen Benennungsoperator für group ByKey ()

groupByKey() -> groupByKey(Grouped.as("kafka-stream-groupby"))

Beispiel für einen Benennungsoperator für normal count ()

normal count() -> .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("kafka-streams-window")  // descriptive name for the store
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long()))

Beispiel für einen Benennungsoperator für Count () im Fenster

windowed count() -> .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("kafka-streams-window")  // descriptive name for the store
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long()))

Beispiel für einen Benennungsoperator für windowed suppressed ()

windowed suppressed() -> 
Suppressed<Windowed> suppressed = Suppressed
        .untilWindowCloses(Suppressed.BufferConfig.unbounded())
        .withName("kafka-suppressed");
    .suppress(suppressed)