Apakah ada cara untuk menghapus semua data dari topik atau menghapus topik sebelum setiap proses?

Dapatkah saya mengubah file KafkaConfig.scala untuk mengubah logRetentionHours properti? Adakah cara agar pesan dihapus segera setelah konsumen membacanya?

Saya menggunakan produsen untuk mengambil data dari suatu tempat dan mengirim data ke topik tertentu di mana konsumen mengkonsumsi, dapatkah saya menghapus semua data dari topik itu di setiap proses? Saya hanya ingin data baru setiap kali dalam topik. Apakah ada cara untuk memulai kembali topik?

Jangan kira itu didukung. Lihat masalah JIRA ini "Tambahkan dukungan penghapusan topik".

Untuk menghapus secara manual:

1. Matikan cluster
2. Bersihkan dir log kafka (ditentukan oleh log.diratribut di file konfigurasi kafka ) serta data penjaga kebun binatang
3. Mulai ulang cluster

Untuk topik tertentu apa yang dapat Anda lakukan adalah

1. Hentikan kafka
2. Bersihkan kafka log khusus untuk partisi, kafka menyimpan file lognya dalam format "logDir / topic-partition" sehingga untuk topik bernama "MyTopic", log untuk partisi id 0 akan disimpan di /tmp/kafka-logs/MyTopic-0tempat /tmp/kafka-logsyang ditentukan oleh log.diratribut
3. Mulai ulang kafka

Ini adalah NOTpendekatan yang bagus dan direkomendasikan tetapi seharusnya berhasil. Di file konfigurasi broker Kafka, log.retention.hours.per.topicatribut digunakan untuk menentukanThe number of hours to keep a log file before deleting it for some specific topic

Selain itu, adakah cara agar pesan dihapus segera setelah konsumen membacanya?

Dari Dokumentasi Kafka :

Kluster Kafka menyimpan semua pesan yang diterbitkan — baik telah dikonsumsi maupun tidak — untuk periode waktu yang dapat dikonfigurasi. Misalnya jika retensi log disetel ke dua hari, maka selama dua hari setelah pesan diterbitkan, pesan tersebut tersedia untuk dikonsumsi, setelah itu akan dibuang untuk mengosongkan ruang. Kinerja Kafka secara efektif konstan sehubungan dengan ukuran data sehingga menyimpan banyak data bukanlah masalah.

Faktanya, satu-satunya metadata yang dipertahankan per konsumen adalah posisi konsumen dalam log, yang disebut "offset". Offset ini dikontrol oleh konsumen: biasanya konsumen akan memajukan offsetnya secara linier saat membaca pesan, tetapi pada kenyataannya posisinya dikontrol oleh konsumen dan dapat menggunakan pesan dalam urutan apa pun yang disukainya. Misalnya konsumen dapat mengatur ulang ke offset yang lebih lama untuk diproses ulang.

Untuk menemukan awal offset untuk membaca di Kafka 0,8 contoh Konsumen Sederhana yang mereka katakan

Kafka menyertakan dua konstanta untuk membantu, kafka.api.OffsetRequest.EarliestTime()menemukan awal data di log dan mulai streaming dari sana, kafka.api.OffsetRequest.LatestTime()hanya akan mengalirkan pesan baru.

Anda juga dapat menemukan kode contoh di sana untuk mengelola offset di ujung konsumen Anda.

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

Seperti yang saya sebutkan di sini Bersihkan Antrian Kafka :

Diuji di Kafka 0.8.2, untuk contoh quick-start: Pertama, Tambahkan satu baris ke file server.properties di bawah folder config:

delete.topic.enable=true

lalu, Anda dapat menjalankan perintah ini:

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

Diuji dengan 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.

Catatan: jika Anda menghapus folder topik di dalam kafka-logs tetapi tidak dari folder zookeeper-data, maka Anda akan melihat topik masih ada.

Sebagai solusi kotor, Anda dapat menyesuaikan pengaturan retensi waktu proses per topik, misalnya bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1( retention.bytes = 0 mungkin juga berfungsi)

Setelah beberapa saat, kafka harus mengosongkan ruang. Tidak yakin apakah ini memiliki implikasi dibandingkan dengan membuat ulang topik.

ps. Lebih baik kembalikan pengaturan retensi, setelah kafka selesai dengan pembersihan.

Anda juga dapat menggunakan retention.msuntuk menyimpan data historis

Di bawah ini adalah script untuk mengosongkan dan menghapus topik Kafka dengan asumsi localhost sebagai server zookeeper dan Kafka_Home diatur ke direktori install:

Skrip di bawah ini akan mengosongkan topik dengan menyetel waktu retensinya ke 1 detik dan kemudian menghapus konfigurasi:

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

Untuk sepenuhnya menghapus topik, Anda harus menghentikan broker kafka yang berlaku dan menghapus direktorinya dari direktori log kafka (default: / tmp / kafka-logs) dan kemudian menjalankan skrip ini untuk menghapus topik dari penjaga kebun binatang. Untuk memverifikasi bahwa itu telah dihapus dari zookeeper, output ls / broker / topik tidak lagi menyertakan topik:

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

Kami mencoba cukup banyak apa yang dijelaskan oleh jawaban lain dengan tingkat keberhasilan yang moderat. Apa yang benar-benar berhasil bagi kami (Apache Kafka 0.8.1) adalah perintah kelas

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

Untuk pengguna minuman

Jika Anda menggunakan brewseperti saya dan membuang banyak waktu untuk mencari kafka-logsfolder yang terkenal itu , jangan takut lagi. (dan tolong beri tahu saya jika itu berhasil untuk Anda dan beberapa versi Homebrew, Kafka dll yang berbeda :))

Anda mungkin akan menemukannya di bawah:

Lokasi:

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

Bagaimana cara menemukan jalan itu

(ini juga berguna untuk pada dasarnya setiap aplikasi yang Anda instal melalui minuman)

1) brew services list

kafka mulai matbhz /Users/matbhz/Library/LaunchAgents/homebrew.mxcl.kafka.plist

2) Buka dan baca yang plistAnda temukan di atas

3) Temukan baris yang menentukan server.propertieslokasi, buka, dalam kasus saya:

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

4) Cari log.dirsbaris:

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

5) Pergi ke lokasi itu dan hapus log untuk topik yang Anda inginkan

6) Mulai ulang Kafka dengan brew services restart kafka

Semua data tentang topik dan partisinya disimpan di tmp/kafka-logs/. Apalagi mereka disimpan dalam sebuah format topic-partionNumber, jadi jika Anda ingin menghapus suatu topik newTopic, Anda dapat:

• hentikan kafka
• hapus file rm -rf /tmp/kafka-logs/newTopic-*

1. Hentikan ZooKeeper dan Kafka
2. Di server.properties, ubah nilai log.retention.hours. Anda dapat mengomentari log.retention.hoursdan menambahkan log.retention.ms=1000. Itu akan membuat rekor di Kafka Topic hanya untuk satu detik.
3. Mulai penjaga kebun binatang dan kafka.
4. Periksa di konsol konsumen. Ketika saya membuka konsol untuk pertama kalinya, ada catatan di sana. Tetapi ketika saya membuka konsol lagi, catatan itu telah dihapus.
5. Nanti, Anda dapat mengatur nilai log.retention.hourske angka yang Anda inginkan.

Pada versi kafka 2.3.0, ada cara alternatif untuk menghapus lunak Kafka (pendekatan lama tidak digunakan lagi).

Perbarui retention.ms menjadi 1 detik (1000ms) lalu setel lagi setelah satu menit, ke pengaturan default yaitu 7 hari (168 jam, 604.800.000 dalam ms)

Penghapusan lunak: - (rentention.ms = 1000) (menggunakan 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'.

Menyetel ke default: - 7 hari (168 jam, 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

Dalam menghapus topik secara manual dari cluster kafka, Anda mungkin saja memeriksanya https://github.com/darrenfu/bigdata/issues/6 Langkah penting yang sering terlewatkan dalam sebagian besar solusi adalah menghapus /config/topics/<topic_name>in ZK.

Saya menggunakan skrip ini:

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

Saya menggunakan utilitas di bawah ini untuk membersihkan setelah uji integrasi saya dijalankan.

Ini menggunakan AdminZkClientapi terbaru . Api lama sudah tidak digunakan lagi.

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)

  }


}

Ada opsi hapus topik. Tapi, itu menandai topik untuk dihapus. Zookeeper kemudian menghapus topik tersebut. Karena ini bisa sangat lama, saya lebih suka pendekatan retention.ms