Kafka Streams DSL sur Kafka Consumer API

Question

Récemment, lors d'un entretien, on m'a posé une question sur Kafka Streams. Plus précisément, l'interviewer voulait savoir pourquoi et quand vous utiliseriez le DSL Kafka Streams plutôt que l'API Kafka Consumer pour lire et traiter des flux de messages ? Je n'ai pas pu fournir de réponse convaincante et je me demande si d'autres personnes qui utilisent ces deux styles de traitement de flux peuvent partager leurs pensées/opinions. Je vous remercie.

Answer 1

Comme d'habitude, cela dépend du cas d'utilisation : quand utiliser l'API KafkaStreams et quand utiliser un simple KafkaProducer/Consumer. Je n'oserais pas choisir l'un plutôt que l'autre en termes généraux.

Tout d'abord, KafkaStreams est construit au-dessus de KafkaProducers/Consumers, de sorte que tout ce qui est possible avec KafkaStreams l'est également avec les simples Consumers/Producers.

Je dirais que l'API KafkaStreams est moins complexe mais aussi moins flexible que les simples Consommateurs/Producteurs. Nous pourrions maintenant entamer de longues discussions sur ce que signifie "moins".

Lorsqu'il s'agit de développer l'API Kafka Streams, vous pouvez directement passer à votre logique d'entreprise en appliquant des méthodes telles que filter , map , join o aggregate car toute la partie consommation et production est abstraite dans les coulisses.

Lorsque vous développez des applications avec des consommateurs/producteurs simples, vous devez réfléchir à la manière dont vous construisez vos clients au niveau de subscribe , poll , send , flush etc.

Si vous voulez encore moins de complexité (mais aussi moins de flexibilité) ksqldb est une autre option que vous pouvez choisir pour construire vos applications Kafka.

Answer 2

Voici quelques scénarios dans lesquels vous pourriez préférer les flux Kafka à l'API de base producteur/consommateur :

1. Il vous permet de construire un pipeline de traitement complexe avec beaucoup de facilité. Supposons que vous ayez un sujet contenant des commandes de clients et que vous souhaitiez filtrer les commandes sur la base d'une ville de livraison et les enregistrer dans une table de la base de données pour la persistance et dans un index Elasticsearch pour une recherche rapide. Dans un tel scénario, vous consommeriez les messages du sujet source, filtreriez les commandes inutiles basées sur la ville en utilisant le DSL Streams filter stocker les données du filtre dans un sujet Kafka séparé (en utilisant la fonction KStream.to() o KTable.to() ), et enfin en utilisant Kafka Connect, les messages seront stockés dans la table de la base de données et dans Elasticsearch. Vous pouvez également faire la même chose en utilisant l'API de base Producer / Consumer, mais cela nécessiterait beaucoup plus de codage.

2. Dans un pipeline de traitement de données, il est possible d'effectuer les opérations de consommation, de traitement et de production au cours d'une même transaction. Ainsi, dans l'exemple ci-dessus, Kafka assurera la sémantique et la transaction exactly-once depuis le sujet source jusqu'à la base de données et Elasticsearch. Il n'y aura pas de messages en double introduits en raison de problèmes de réseau et de tentatives. Cette fonctionnalité est particulièrement utile lorsque vous effectuez des agrégats tels que le nombre de commandes au niveau d'un produit individuel. Dans de tels scénarios, les doublons vous donneront toujours des résultats erronés.

3. Vous pouvez également enrichir vos données entrantes avec une latence très faible. Supposons que, dans l'exemple ci-dessus, vous souhaitiez enrichir les données de la commande avec l'adresse électronique du client à partir de vos données clients stockées. En l'absence de Kafka Streams, que feriez-vous ? Vous invoqueriez probablement une API REST pour chaque commande entrante sur le réseau, ce qui serait certainement une opération coûteuse ayant un impact sur votre débit. Dans ce cas, vous pourriez stocker les données clients requises dans un sujet Kafka compacté et les charger dans l'application de streaming à l'aide de la fonction KTable o GlobalKTable . Il ne vous reste plus qu'à effectuer une simple recherche locale dans la table KT pour l'adresse électronique du client. Notez que les données de la table KT seront stockées dans la base de données RocksDB intégrée qui est fournie avec Kafka Streams et, comme la table KT est soutenue par un sujet Kafka, vos données dans l'application de streaming seront continuellement mises à jour en temps réel. En d'autres termes, il n'y aura pas de données périmées. Il s'agit essentiellement d'un exemple de vue matérialisée.

4. Supposons que vous souhaitiez joindre deux flux de données différents. Ainsi, dans l'exemple ci-dessus, vous souhaitez traiter uniquement les commandes dont le paiement a été effectué avec succès et les données de paiement proviennent d'un autre sujet Kafka. Il peut arriver que le paiement soit retardé ou que l'événement de paiement précède l'événement de commande. Dans ce cas, vous voudrez peut-être effectuer une jointure fenêtrée d'une heure. Ainsi, si la commande et les événements de paiement correspondants interviennent dans une fenêtre d'une heure, la commande sera autorisée à poursuivre son traitement dans le pipeline. Comme vous pouvez le constater, vous devez stocker l'état pour une fenêtre d'une heure et cet état sera stocké dans la base de données Rocks des flux Kafka.