Apache Kafka - KafkaStream auf Thema/partition

Ich Schreibe Kafka Verbraucher für high volume, high velocity, verteilte Anwendung. Ich habe nur ein Thema, sondern rate der eingehenden Nachrichten ist sehr hoch. Mit mehreren partition, dienen mehr Verbraucher wäre angemessen für diesen use-case. Beste Weg, Sie zu konsumieren haben mehrere stream-Leser. Nach den Unterlagen oder Proben zur Verfügung, die Anzahl der KafkaStreams die ConsumerConnector gibt basiert auf der Anzahl der Themen. Fragen, wie man mehr als eine KafkaStream Leser [auf der Grundlage der partition], so dass ich nicht unbedingt ein thread pro stream oder Lesen aus der gleichen KafkaStream in mehreren threads tun würde, das gleichzeitige Lesen von mehreren Partitionen?

Alle Erkenntnisse sind sehr geschätzt.

  • Mit SimpleConsumer ist nicht eine option?
InformationsquelleAutor Nageswara Rao | 2014-05-11
  1. 15

    Möchte teilen, was ich gefunden von der mailing-Liste:

    Die Zahl, die Sie übergeben in die topic-map-Steuerelemente, wie viele Datenströme ein Thema ist unterteilt in. In deinem Fall, wenn du in 1 übergeben, werden alle 10 Partitionen, die die Daten eingespeist werden 1 stream. Wenn Sie 2, jede der 2-streams werden die Daten aus 5 Partitionen. Wenn Sie übergeben, 11, 10 von Ihnen erhalten Daten aus 1 Trennwand und 1 stream wird nichts bekommen.

    In der Regel müssen Sie Durchlaufen jeden stream in einem eigenen thread. Dies ist, weil jeder stream kann immer blockieren, wenn es kein neues Ereignis.

    Beispiel-snippet:

    topicCount.put(msgTopic, new Integer(partitionCount));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = connector.createMessageStreams(topicCount);
List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(msgTopic);

for (final KafkaStream stream : streams) {
    ReadTask task = new ReadTask(stream, msgTopic);
    task.addObserver(this.msgObserver);
    tasks.add(task); executor.submit(task);
}

    Referenz: http://mail-archives.apache.org/mod_mbox/incubator-kafka-users/201201.mbox/%3CCA+sHyy_Z903dOmnjp7_yYR_aE2sRW-x7XpAnqkmWaP66GOqf6w@mail.gmail.com%3E

    • Beispiel-snippet topicCount.put(msgTopic, new Integer(partitionCount)); Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = connector.createMessageStreams(topicCount); Liste<KafkaStream<byte[], byte[]>> streams = consumerStreams.bekommen(msgTopic); for (final KafkaStream stream : Stream) { ReadTask task = new ReadTask(Strom, msgTopic); task.addObserver(dies.msgObserver); Aufgaben.add(Aufgabe); executor.submit(task); }
    InformationsquelleAutor Nageswara Rao
  2. 4

    Ist der empfohlene Weg, dies zu tun ist, um eine thread-pool-also Java kann die organisation für Sie und für jeden stream die createMessageStreamsByFilter Methode gibt, die Sie verbrauchen, in eine Runnable. Zum Beispiel:

    int NUMBER_OF_PARTITIONS = 6;
Properties consumerConfig = new Properties();
consumerConfig.put("zk.connect", "zookeeper.mydomain.com:2181" );
consumerConfig.put("backoff.increment.ms", "100");
consumerConfig.put("autooffset.reset", "largest");
consumerConfig.put("groupid", "java-consumer-example");
consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerConfig));

TopicFilter sourceTopicFilter = new Whitelist("mytopic|myothertopic");
List<KafkaStream<Message>> streams = consumer.createMessageStreamsByFilter(sourceTopicFilter, NUMBER_OF_PARTITIONS);

ExecutorService executor = Executors.newFixedThreadPool(streams.size());
for(final KafkaStream<Message> stream: streams){
    executor.submit(new Runnable() {
        public void run() {
            for (MessageAndMetadata<Message> msgAndMetadata: stream) {
                ByteBuffer buffer = msgAndMetadata.message().payload();
                byte [] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                //Do something with the bytes you just got off Kafka.
            }
        }
    });
}

    In diesem Beispiel bat ich um 6 threads Grunde, weil ich weiß, dass ich 3 Partitionen für jedes Thema und ich zählte zwei Themen, die in meiner whitelist. Einmal haben wir die Griffe des eingehenden streams können wir iterieren über Ihre Inhalte, die MessageAndMetadata Objekte. Metadaten ist wirklich nur das Thema Namen und offset. Als Sie entdeckt, Sie können es in einem einzigen thread, wenn Sie Fragen, für 1-stream statt, in meinem Beispiel 6, aber wenn Sie benötigen, parallele Verarbeitung die feine englische Art ist, starten Sie einen Testamentsvollstrecker mit einem thread für jede zurückgegebene stream.

    • Was würde passieren, wenn ich dies tun? kafkaConsumerConfig = new ConsumerConfig(...); consumerConnector = Verbraucher.createJavaConsumerConnector(kafkaConsumerConfig); topicCountMap.put("mytopic", 1); consumerMap.get("mytopic").get(0); Überprüfen Sie es, ist get(0) auf, die Liste der kafka-streams, so bekomme ich nur 1 stream. Was passiert, wenn ich call-Verbraucher.createJavaConsumerConnector 10 mal?
    • Würden Sie alle teilen die gleiche config und jeder wird das Lesen aller Partitionen, so dass ich denke, würde man 10 Verbrauchern, von denen alle würden versuchen, speichern Sie Ihre Staat in der gleichen ZK-Knoten, so dass Sie würde am Ende mit Verbraucher 1 z.B. das Lesen der ersten 1K-Nachrichten, dann Verbraucher 2 Lesen die gleichen 1K-Nachrichten, aber potenziell Verbraucher 1 fertig würde, das Lesen der batch-update ZK, ein zweites Lesen, schreiben Sie dann Ihren Standpunkt ZK dann unsere zweite aus irgendeinem Grund langsamer thread kommt und schreibt seine position zurück zu ZK, die die erste consumer-re-Verarbeitung der zweiten charge. Grundsätzlich Konflikte in Hülle und fülle.
    InformationsquelleAutor feldoh
  3. 1
    /**
 * @param source : source kStream to sink output-topic
 */
private static void pipe(KStream<String, String> source) {
    source.to(Serdes.String(), Serdes.String(), new StreamPartitioner<String, String>() {

        @Override
        public Integer partition(String arg0, String arg1, int arg2) {
            return 0;
        }
    }, "output-topic");
}

    obige code schreiben, Datensatz in der partition 1 der topic-name "- Ausgabe-Thema"

    InformationsquelleAutor Abhinav
Schreibe einen Kommentar