Existe-t-il un moyen de supprimer toutes les données d'une rubrique ou de supprimer la rubrique avant chaque exécution?

Puis-je modifier le fichier KafkaConfig.scala pour changer la logRetentionHourspropriété? Existe-t-il un moyen de supprimer les messages dès que le consommateur les lit?

J'utilise des producteurs pour récupérer les données de quelque part et envoyer les données à un sujet particulier où un consommateur consomme, puis-je supprimer toutes les données de ce sujet à chaque exécution? Je ne veux que de nouvelles données à chaque fois dans le sujet. Existe-t-il un moyen de réinitialiser le sujet d'une manière ou d'une autre?

Ne pensez pas qu'il est encore pris en charge. Jetez un œil à ce problème JIRA "Ajout de la prise en charge de la suppression de sujets".

Pour supprimer manuellement:

1. Arrêter le cluster
2. Nettoyer le répertoire du journal kafka (spécifié par l' log.dirattribut dans le fichier de configuration kafka ) ainsi que les données du gardien de zoo
3. Redémarrez le cluster

Pour un sujet donné, ce que vous pouvez faire est

1. Arrêtez kafka
2. Nettoyer le journal kafka spécifique à la partition, kafka stocke son fichier journal dans un format de «logDir / topic-partition» donc pour une rubrique nommée «MyTopic», le journal de la partition id 0 sera stocké /tmp/kafka-logs/MyTopic-0là où /tmp/kafka-logsest spécifié par l' log.dirattribut
3. Redémarrez kafka

C'est NOTune bonne approche recommandée, mais elle devrait fonctionner. Dans le fichier de configuration du courtier Kafka, l' log.retention.hours.per.topicattribut est utilisé pour définirThe number of hours to keep a log file before deleting it for some specific topic

En outre, existe-t-il un moyen de supprimer les messages dès que le consommateur les lit?

À partir de la documentation Kafka :

Le cluster Kafka conserve tous les messages publiés, qu'ils aient été consommés ou non, pendant une période configurable. Par exemple, si la rétention du journal est définie sur deux jours, pendant les deux jours suivant la publication d'un message, il est disponible à la consommation, après quoi il sera supprimé pour libérer de l'espace. Les performances de Kafka sont effectivement constantes en ce qui concerne la taille des données, donc la conservation d'un grand nombre de données n'est pas un problème.

En fait, les seules métadonnées conservées par consommateur sont la position du consommateur dans le journal, appelée «offset». Ce décalage est contrôlé par le consommateur: normalement, un consommateur avance son décalage de manière linéaire lorsqu'il lit les messages, mais en fait, la position est contrôlée par le consommateur et il peut consommer les messages dans l'ordre de son choix. Par exemple, un consommateur peut revenir à un décalage plus ancien pour le retraiter.

Pour trouver le décalage de début à lire dans Kafka 0.8 Exemple de consommateur simple, ils disent

Kafka inclut deux constantes pour vous aider, kafka.api.OffsetRequest.EarliestTime()trouve le début des données dans les journaux et commence à diffuser à partir de là, kafka.api.OffsetRequest.LatestTime()ne diffusera que les nouveaux messages.

Vous pouvez également y trouver l'exemple de code pour gérer l'offset chez votre consommateur.

    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition,
                                 long whichTime, String clientName) {
    TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
    Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
    requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
    kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(),clientName);
    OffsetResponse response = consumer.getOffsetsBefore(request);

    if (response.hasError()) {
        System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition) );
        return 0;
    }
    long[] offsets = response.offsets(topic, partition);
    return offsets[0];
}

Comme je l'ai mentionné ici Purge Kafka Queue :

Testé dans Kafka 0.8.2, pour l'exemple de démarrage rapide: Tout d'abord, ajoutez une ligne au fichier server.properties sous le dossier config:

delete.topic.enable=true

ensuite, vous pouvez exécuter cette commande:

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test

Testé avec kafka 0.10

1. stop zookeeper & Kafka server,
2. then go to 'kafka-logs' folder , there you will see list of kafka topic folders, delete folder with topic name
3. go to 'zookeeper-data' folder , delete data inside that.
4. start zookeeper & kafka server again.

Remarque: si vous supprimez le (s) dossier (s) de sujets dans kafka-logs mais pas du dossier zookeeper-data, vous verrez que les sujets sont toujours là.

Pour contourner le problème, vous pouvez ajuster les paramètres de rétention d'exécution par sujet, par exemple bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1( retention.bytes = 0 peut également fonctionner)

Après un court instant, kafka devrait libérer de l'espace. Je ne sais pas si cela a des implications par rapport à la recréation du sujet.

ps. Mieux vaut ramener les paramètres de rétention, une fois kafka terminé avec le nettoyage.

Vous pouvez également utiliser retention.mspour conserver les données historiques

Vous trouverez ci-dessous des scripts pour vider et supprimer une rubrique Kafka en supposant que localhost est le serveur zookeeper et que Kafka_Home est défini sur le répertoire d'installation:

Le script ci-dessous videra un sujet en définissant son temps de rétention sur 1 seconde, puis en supprimant la configuration:

#!/bin/bash
echo "Enter name of topic to empty:"
read topicName
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --add-config retention.ms=1000
sleep 5
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --delete-config retention.ms

Pour supprimer complètement les sujets, vous devez arrêter tous les courtiers kafka applicables et supprimer son ou ses répertoires du répertoire du journal kafka (par défaut: / tmp / kafka-logs), puis exécuter ce script pour supprimer le sujet de zookeeper. Pour vérifier qu'il a été supprimé de zookeeper, la sortie de ls / brokers / topics ne doit plus inclure le sujet:

#!/bin/bash
echo "Enter name of topic to delete from zookeeper:"
read topicName
/$Kafka_Home/bin/zookeeper-shell localhost:2181 <<EOF
rmr /brokers/topics/$topicName
ls /brokers/topics
quit
EOF

Nous avons essayé à peu près ce que les autres réponses décrivent avec un niveau de succès modéré. Ce qui a vraiment fonctionné pour nous (Apache Kafka 0.8.1) est la commande de classe

sh kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic yourtopic --zookeeper localhost: 2181

Pour les utilisateurs de bière

Si vous utilisez brewcomme moi et que vous avez perdu beaucoup de temps à chercher le fameux kafka-logsdossier, n'ayez plus peur. (et faites-le moi savoir si cela fonctionne pour vous et pour plusieurs versions différentes de Homebrew, Kafka, etc. :))

Vous allez probablement le trouver sous:

Emplacement:

/usr/local/var/lib/kafka-logs

Comment trouver réellement ce chemin

(cela est également utile pour pratiquement toutes les applications que vous installez via brew)

1) brew services list

kafka a démarré matbhz /Users/matbhz/Library/LaunchAgents/homebrew.mxcl.kafka.plist

2) Ouvrez et lisez ce que plistvous avez trouvé ci-dessus

3) Trouvez la ligne définissant l' server.propertiesemplacement, ouvrez-la, dans mon cas:

• /usr/local/etc/kafka/server.properties

4) Recherchez la log.dirsligne:

log.dirs = / usr / local / var / lib / kafka-logs

5) Allez à cet emplacement et supprimez les journaux des sujets que vous souhaitez

6) Redémarrez Kafka avec brew services restart kafka

Toutes les données sur les sujets et ses partitions sont stockées dans tmp/kafka-logs/. De plus, ils sont stockés dans un format topic-partionNumber, donc si vous souhaitez supprimer un sujet newTopic, vous pouvez:

• arrête kafka
• supprimer les fichiers rm -rf /tmp/kafka-logs/newTopic-*

1. Arrêtez ZooKeeper et Kafka
2. Dans server.properties, modifiez la valeur log.retention.hours. Vous pouvez commenter log.retention.hourset ajouter log.retention.ms=1000. Il garderait le dossier sur Kafka Topic pendant une seconde seulement.
3. Démarrez le gardien de zoo et kafka.
4. Vérifiez sur la console du consommateur. Quand j'ai ouvert la console pour la première fois, le disque était là. Mais quand j'ai rouvert la console, l'enregistrement a été supprimé.
5. Plus tard, vous pouvez définir la valeur de log.retention.hourssur la valeur souhaitée.

À partir de la version kafka 2.3.0, il existe un autre moyen de supprimer en douceur Kafka (l'ancienne approche est obsolète).

Mettez à jour retention.ms à 1 sec (1000ms) puis réglez-le à nouveau après une minute, au paramètre par défaut, c'est-à-dire 7 jours (168 heures, 604 800 000 en ms)

Suppression logicielle : - (rentention.ms = 1000) (en utilisant kafka-configs.sh)

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=1000
Completed Updating config for entity: topic 'kafka_topic3p3r'.

Valeur par défaut: - 7 jours (168 heures, retention.ms = 604800000)

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=604800000

En supprimant manuellement un sujet d'un cluster kafka, vous pouvez simplement vérifier ceci https://github.com/darrenfu/bigdata/issues/6 Une étape vitale manquée dans la plupart des solutions consiste à supprimer le /config/topics/<topic_name>dans ZK.

J'utilise ce script:

#!/bin/bash
topics=`kafka-topics --list --zookeeper zookeeper:2181`
for t in $topics; do 
    for p in retention.ms retention.bytes segment.ms segment.bytes; do
        kafka-topics --zookeeper zookeeper:2181 --alter --topic $t --config ${p}=100
    done
done
sleep 60
for t in $topics; do 
    for p in retention.ms retention.bytes segment.ms segment.bytes; do
        kafka-topics --zookeeper zookeeper:2181 --alter --topic $t --delete-config ${p}
    done
done

J'utilise l'utilitaire ci-dessous pour nettoyer après mon test d'intégration.

Il utilise la dernière AdminZkClientapi. L'ancienne API est obsolète.

import javax.inject.Inject
import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.common.utils.Time

class ZookeeperUtils @Inject() (config: AppConfig) {

  val testTopic = "users_1"

  val zkHost = config.KafkaConfig.zkHost
  val sessionTimeoutMs = 10 * 1000
  val connectionTimeoutMs = 60 * 1000
  val isSecure = false
  val maxInFlightRequests = 10
  val time: Time = Time.SYSTEM

  def cleanupTopic(config: AppConfig) = {

    val zkClient = KafkaZkClient.apply(zkHost, isSecure, sessionTimeoutMs, connectionTimeoutMs, maxInFlightRequests, time)
    val zkUtils = new AdminZkClient(zkClient)

    val pp = new Properties()
    pp.setProperty("delete.retention.ms", "10")
    pp.setProperty("file.delete.delay.ms", "1000")
    zkUtils.changeTopicConfig(testTopic , pp)
    //    zkUtils.deleteTopic(testTopic)

    println("Waiting for topic to be purged. Then reset to retain records for the run")
    Thread.sleep(60000L)

    val resetProps = new Properties()
    resetProps.setProperty("delete.retention.ms", "3000000")
    resetProps.setProperty("file.delete.delay.ms", "4000000")
    zkUtils.changeTopicConfig(testTopic , resetProps)

  }


}

Il existe une option de suppression de sujet. Mais, cela marque le sujet de la suppression. Zookeeper supprime plus tard le sujet. Comme cela peut être d'une durée imprévisible, je préfère l'approche retention.ms