Czy istnieje sposób na usunięcie wszystkich danych z tematu lub usunięcie tematu przed każdym uruchomieniem?

Czy mogę zmodyfikować plik KafkaConfig.scala, aby zmienić logRetentionHourswłaściwość? Czy istnieje sposób, w jaki wiadomości są usuwane, gdy tylko konsument je przeczyta?

Używam producentów do pobierania danych skądś i wysyłania danych do określonego tematu, w którym konsument konsumuje, czy mogę usunąć wszystkie dane z tego tematu przy każdym uruchomieniu? Chcę tylko nowych danych za każdym razem w temacie. Czy jest sposób na ponowne zainicjowanie tematu?

Nie myśl, że jest jeszcze obsługiwany. Spójrz na to wydanie JIRA „Dodaj obsługę usuwania tematów”.

Aby usunąć ręcznie:

1. Zamknij klaster
2. Wyczyść katalog dziennika kafka (określony przez log.diratrybut w pliku konfiguracyjnym kafka ) oraz dane zookeepera
3. Uruchom ponownie klaster

W przypadku każdego tematu możesz zrobić

1. Przestań kafka
2. Wyczyść dziennik kafka specyficzny dla partycji, kafka przechowuje swój plik dziennika w formacie „logDir / topic-partition”, więc dla tematu o nazwie „MyTopic” dziennik partycji o identyfikatorze 0 będzie przechowywany w /tmp/kafka-logs/MyTopic-0miejscu /tmp/kafka-logsokreślonym przez log.diratrybut
3. Uruchom ponownie kafka

To jest NOTdobre i zalecane podejście, ale powinno działać. W pliku konfiguracyjnym brokera Kafka log.retention.hours.per.topicatrybut jest używany do definiowaniaThe number of hours to keep a log file before deleting it for some specific topic

Czy istnieje również sposób, w jaki wiadomości są usuwane, gdy tylko konsument je przeczyta?

Z dokumentacji Kafki :

Klaster Kafka zachowuje wszystkie opublikowane wiadomości - niezależnie od tego, czy zostały zużyte - przez konfigurowalny okres czasu. Na przykład, jeśli czas przechowywania dziennika jest ustawiony na dwa dni, to przez dwa dni po opublikowaniu wiadomości jest on dostępny do użytku, po czym zostanie wyrzucony w celu zwolnienia miejsca. Wydajność Kafki jest w rzeczywistości stała w odniesieniu do rozmiaru danych, więc przechowywanie dużej ilości danych nie stanowi problemu.

W rzeczywistości jedynymi metadanymi zachowanymi dla każdego konsumenta jest pozycja konsumenta w dzienniku, zwana „przesunięciem”. To przesunięcie jest kontrolowane przez konsumenta: zwykle konsument przesuwa swoje przesunięcie liniowo podczas odczytywania wiadomości, ale w rzeczywistości pozycja jest kontrolowana przez konsumenta i może konsumować wiadomości w dowolnej kolejności. Na przykład konsument może zresetować do starszego przesunięcia w celu ponownego przetworzenia.

Mówią, że za znalezienie przesunięcia początkowego do przeczytania w przykładzie Prostego konsumenta Kafki 0.8

Kafka zawiera dwie stałe, które mogą pomóc, kafka.api.OffsetRequest.EarliestTime()znajduje początek danych w dziennikach i rozpoczyna strumieniowanie od tego miejsca, kafka.api.OffsetRequest.LatestTime()będzie przesyłał strumieniowo tylko nowe wiadomości.

Możesz tam również znaleźć przykładowy kod do zarządzania przesunięciem po stronie klienta.

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

Jak wspomniałem tutaj Oczyść kolejkę Kafki :

Testowane w Kafka 0.8.2, dla przykładu szybkiego startu: Najpierw dodaj jedną linię do pliku server.properties w folderze konfiguracyjnym:

delete.topic.enable=true

następnie możesz uruchomić to polecenie:

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

Testowany z kafką 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.

Uwaga: jeśli usuwasz folder / y tematów wewnątrz dzienników kafka, ale nie z folderu zookeeper-data, zobaczysz, że tematy nadal tam są.

Jako brudne obejście można dostosować ustawienia przechowywania w czasie wykonywania dla poszczególnych tematów, np. bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1( Może również działać retention.bytes = 0 )

Po chwili kafka powinna zwolnić miejsce. Nie jestem pewien, czy ma to jakieś konsekwencje w porównaniu z ponownym utworzeniem tematu.

ps. Lepiej przywróć ustawienia retencji, gdy kafka zakończy czyszczenie.

Możesz również użyć retention.msdo utrwalenia danych historycznych

Poniżej znajdują się skrypty do opróżniania i usuwania tematu Kafki, przyjmując localhost jako serwer zookeeper, a Kafka_Home jest ustawiony na katalog instalacyjny:

Poniższy skrypt opróżni temat, ustawiając jego czas przechowywania na 1 sekundę, a następnie usuwając konfigurację:

#!/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

Aby całkowicie usunąć tematy, należy zatrzymać wszystkie odpowiednie brokery kafka i usunąć jego katalog (y) z katalogu dziennika kafka (domyślnie: / tmp / kafka-logs), a następnie uruchomić ten skrypt, aby usunąć temat z zookeepera. Aby sprawdzić, czy został usunięty z zookeepera, dane wyjściowe ls / brokers / topics nie powinny już zawierać tematu:

#!/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

Wypróbowaliśmy mniej więcej to, co opisują inne odpowiedzi, z umiarkowanym poziomem sukcesu. To, co naprawdę zadziałało dla nas (Apache Kafka 0.8.1), to polecenie klasy

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

Dla użytkowników piwa

Jeśli używasz czegoś brewtakiego jak ja i tracisz dużo czasu na szukanie niesławnego kafka-logsfolderu, nie bój się więcej. (i daj mi znać, jeśli to działa dla Ciebie i wielu różnych wersji Homebrew, Kafka itp. :))

Prawdopodobnie znajdziesz to pod:

Lokalizacja:

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

Jak właściwie znaleźć tę ścieżkę

(jest to również przydatne w zasadzie dla każdej aplikacji instalowanej przez brew)

1) brew services list

kafka rozpoczął matbhz /Users/matbhz/Library/LaunchAgents/homebrew.mxcl.kafka.plist

2) Otwórz i przeczytaj, plistże znalazłeś powyżej

3) Znajdź linię określającą server.propertieslokalizację otwórz ją, w moim przypadku:

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

4) Poszukaj log.dirslinii:

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

5) Przejdź do tej lokalizacji i usuń dzienniki dotyczące żądanych tematów

6) Zrestartuj Kafkę za pomocą brew services restart kafka

Wszystkie dane dotyczące tematów i ich partycji są przechowywane w plikach tmp/kafka-logs/. Ponadto są przechowywane w formacie topic-partionNumber, więc jeśli chcesz usunąć temat newTopic, możesz:

• przestań kafka
• usuń pliki rm -rf /tmp/kafka-logs/newTopic-*

1. Zatrzymaj ZooKeeper i Kafka
2. W pliku server.properties zmień wartość log.retention.hours. Możesz komentować log.retention.hoursi dodawać log.retention.ms=1000. To zachowałoby rekord Kafki Topic tylko przez jedną sekundę.
3. Uruchom zookeeper i kafka.
4. Sprawdź w konsoli klienta. Kiedy otworzyłem konsolę po raz pierwszy, był tam rekord. Ale kiedy ponownie otworzyłem konsolę, płyta została usunięta.
5. Później możesz ustawić wartość log.retention.hoursna żądaną figurę.

Od wersji kafka 2.3.0 istnieje alternatywny sposób miękkiego usuwania Kafki (stare podejście jest przestarzałe).

Zaktualizuj retention.ms do 1 sekundy (1000 ms), a następnie ustaw ponownie po minucie, do ustawienia domyślnego, tj. 7 dni (168 godzin, 604 800 000 w ms)

Miękkie usuwanie: - (rentention.ms = 1000) (przy użyciu 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'.

Ustawienie domyślne: - 7 dni (168 godzin, 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

Podczas ręcznego usuwania tematu z klastra kafka, możesz po prostu sprawdzić to https://github.com/darrenfu/bigdata/issues/6 Istotnym krokiem, którego wiele brakuje w większości rozwiązań, jest usunięcie /config/topics/<topic_name>w ZK.

Używam tego skryptu:

#!/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

Używam poniższego narzędzia do czyszczenia po uruchomieniu testu integracji.

Korzysta z najnowszego AdminZkClientinterfejsu API. Starszy interfejs API został wycofany.

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)

  }


}

Istnieje opcja usunięcia tematu. Ale oznacza temat do usunięcia. Zookeeper później usuwa temat. Ponieważ może to być nieprzewidywalnie długie, wolę podejście retention.ms