¿Hay alguna forma de eliminar todos los datos de un tema o eliminar el tema antes de cada ejecución?

¿Puedo modificar el archivo KafkaConfig.scala para cambiar la logRetentionHourspropiedad? ¿Hay alguna forma de que los mensajes se eliminen tan pronto como el consumidor los lea?

Estoy usando productores para obtener los datos de algún lugar y enviar los datos a un tema en particular donde un consumidor consume, ¿puedo eliminar todos los datos de ese tema en cada ejecución? Quiero solo datos nuevos cada vez en el tema. ¿Hay alguna forma de reinicializar el tema de alguna manera?

No creo que sea compatible todavía. Eche un vistazo a este número de JIRA "Agregar soporte para eliminar temas".

Para eliminar manualmente:

1. Apaga el clúster
2. Limpiar el directorio de registro de kafka (especificado por el log.diratributo en el archivo de configuración de kafka ) así como los datos del guardián del zoológico
3. Reinicie el clúster

Para cualquier tema dado, lo que puede hacer es

1. Detener kafka
2. Limpiar el registro de kafka específico para la partición, kafka almacena su archivo de registro en un formato de "logDir / topic-partition", por lo que para un tema llamado "MyTopic", el registro de la partición id 0 se almacenará en el /tmp/kafka-logs/MyTopic-0lugar /tmp/kafka-logsespecificado por el log.diratributo
3. Reiniciar kafka

Este es NOTun enfoque bueno y recomendado, pero debería funcionar. En el archivo de configuración del corredor de Kafka, el log.retention.hours.per.topicatributo se usa para definirThe number of hours to keep a log file before deleting it for some specific topic

Además, ¿hay alguna forma de que los mensajes se eliminen tan pronto como el consumidor los lea?

De la documentación de Kafka :

El clúster de Kafka retiene todos los mensajes publicados, ya sea que se hayan consumido o no, durante un período de tiempo configurable. Por ejemplo, si la retención de registros se establece en dos días, durante los dos días posteriores a la publicación de un mensaje, estará disponible para el consumo, después de lo cual se descartará para liberar espacio. El rendimiento de Kafka es efectivamente constante con respecto al tamaño de los datos, por lo que retener muchos datos no es un problema.

De hecho, los únicos metadatos que se conservan por consumidor es la posición del consumidor en el registro, denominada "compensación". Este desplazamiento es controlado por el consumidor: normalmente un consumidor avanzará su desplazamiento linealmente a medida que lee los mensajes, pero de hecho la posición está controlada por el consumidor y puede consumir mensajes en el orden que desee. Por ejemplo, un consumidor puede restablecer un desplazamiento anterior para reprocesar.

Para encontrar el desplazamiento inicial para leer en el ejemplo de consumidor simple de Kafka 0.8 , dicen

Kafka incluye dos constantes para ayudar, kafka.api.OffsetRequest.EarliestTime()encuentra el comienzo de los datos en los registros y comienza a transmitir desde allí, kafka.api.OffsetRequest.LatestTime()solo transmitirá mensajes nuevos.

También puede encontrar el código de ejemplo allí para administrar la compensación en el extremo de su consumidor.

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

Como mencioné aquí Purge Kafka Queue :

Probado en Kafka 0.8.2, para el ejemplo de inicio rápido: Primero, agregue una línea al archivo server.properties en la carpeta de configuración:

delete.topic.enable=true

entonces, puede ejecutar este comando:

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

Probado con 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.

Nota: si está eliminando carpetas de temas dentro de kafka-logs pero no de la carpeta zookeeper-data, verá que los temas siguen ahí.

Como solución sucia, puede ajustar la configuración de retención de tiempo de ejecución por tema, por ejemplo bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1( retención.bytes = 0 también podría funcionar)

Después de un rato, kafka debería liberar espacio. No estoy seguro si esto tiene alguna implicación en comparación con la recreación del tema.

PD. Es mejor recuperar la configuración de retención, una vez que Kafka haya terminado con la limpieza.

También puede utilizar retention.mspara conservar datos históricos

A continuación se muestran los scripts para vaciar y eliminar un tema de Kafka asumiendo que localhost como el servidor del guardián del zoológico y Kafka_Home está configurado en el directorio de instalación:

El siguiente script vaciará un tema estableciendo su tiempo de retención en 1 segundo y luego eliminando la configuración:

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

Para eliminar completamente los temas, debe detener cualquier corredor de kafka aplicable y eliminar sus directorios del directorio de registro de kafka (predeterminado: / tmp / kafka-logs) y luego ejecutar este script para eliminar el tema de zookeeper. Para verificar que se haya eliminado de zookeeper, la salida de ls / brokers / topics ya no debería incluir el tema:

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

Intentamos prácticamente lo que describen las otras respuestas con un nivel moderado de éxito. Lo que realmente funcionó para nosotros (Apache Kafka 0.8.1) es el comando de clase

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

Para usuarios de cerveza

Si está usando brewcomo yo y perdió mucho tiempo buscando la kafka-logscarpeta infame , no temas más. (y hágamelo saber si eso funciona para usted y varias versiones diferentes de Homebrew, Kafka, etc. :))

Probablemente lo encontrará en:

Ubicación:

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

Cómo encontrar realmente ese camino

(esto también es útil para básicamente todas las aplicaciones que instala a través de brew)

1) brew services list

kafka inició matbhz /Users/matbhz/Library/LaunchAgents/homebrew.mxcl.kafka.plist

2) Abra y lea lo que plistencontró arriba

3) Encuentra la línea que define la server.propertiesubicación, ábrela, en mi caso:

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

4) Busque la log.dirslínea:

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

5) Vaya a esa ubicación y elimine los registros de los temas que desee

6) Reinicie Kafka con brew services restart kafka

Todos los datos sobre temas y sus particiones se almacenan en tmp/kafka-logs/. Además, están almacenados en un formato topic-partionNumber, por lo que si desea eliminar un tema newTopic, puede:

• detener kafka
• borrar los archivos rm -rf /tmp/kafka-logs/newTopic-*

1. Detén a ZooKeeper y Kafka
2. En server.properties, cambie el valor de log.retention.hours. Puedes comentar log.retention.hoursy agregar log.retention.ms=1000. Mantendría el récord de Kafka Topic por solo un segundo.
3. Empiece a cuidar del zoológico y kafka.
4. Verifique en la consola del consumidor. Cuando abrí la consola por primera vez, el registro estaba allí. Pero cuando volví a abrir la consola, el registro se eliminó.
5. Más adelante, puede establecer el valor de log.retention.hoursla cifra deseada.

A partir de la versión de kafka 2.3.0, existe una forma alternativa de eliminación suave de Kafka (el enfoque antiguo está en desuso).

Actualice la retención.ms a 1 segundo (1000 ms) y luego configúrelo nuevamente después de un minuto, a la configuración predeterminada, es decir, 7 días (168 horas, 604,800,000 en ms)

Eliminación suave: - (rentention.ms = 1000) (usando 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'.

Configuración predeterminada: - 7 días (168 horas, retención.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

Al eliminar manualmente un tema de un clúster de kafka, puede consultar https://github.com/darrenfu/bigdata/issues/6 Un paso vital que se perdió mucho en la mayoría de las soluciones es eliminar el /config/topics/<topic_name>archivo en ZK.

Yo uso este 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

Utilizo la utilidad a continuación para limpiar después de mi ejecución de prueba de integración.

Utiliza lo último AdminZkClient API. La API anterior ha quedado obsoleta.

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)

  }


}

Hay una opción para eliminar tema. Pero marca el tema para su eliminación. Zookeeper luego elimina el tema. Dado que esto puede ser impredeciblemente largo, prefiero el enfoque Hold.ms