Existe uma maneira de excluir todos os dados de um tópico ou excluir o tópico antes de cada execução?

Posso modificar o arquivo KafkaConfig.scala para alterar a logRetentionHourspropriedade? Existe uma maneira de as mensagens serem excluídas assim que o consumidor as lê?

Estou usando produtores para buscar os dados de algum lugar e enviando os dados para um tópico específico onde um consumidor consome. Posso excluir todos os dados desse tópico em cada execução? Quero apenas novos dados a cada vez no tópico. Existe uma maneira de reinicializar o tópico de alguma forma?

Não acho que seja compatível ainda. Dê uma olhada neste problema do JIRA "Adicionar suporte para exclusão de tópicos".

Para excluir manualmente:

1. Desligue o cluster
2. Limpe o diretório de log kafka (especificado pelo log.diratributo no arquivo de configuração kafka ), bem como os dados do zookeeper
3. Reinicie o cluster

Para qualquer tópico, o que você pode fazer é

1. Pare de kafka
2. Limpe o log kafka específico da partição, o kafka armazena seu arquivo de log em um formato de "logDir / topic-partition" para um tópico denominado "MyTopic", o log para o id da partição 0 será armazenado /tmp/kafka-logs/MyTopic-0onde /tmp/kafka-logsé especificado pelo log.diratributo
3. Reiniciar kafka

Esta é NOTuma abordagem boa e recomendada, mas deve funcionar. No arquivo de configuração do corretor Kafka, o log.retention.hours.per.topicatributo é usado para definirThe number of hours to keep a log file before deleting it for some specific topic

Além disso, existe uma maneira de as mensagens serem excluídas assim que o consumidor as lê?

Da documentação do Kafka :

O cluster Kafka retém todas as mensagens publicadas - quer tenham ou não sido consumidas - por um período de tempo configurável. Por exemplo, se a retenção de log for definida para dois dias, nos dois dias após a publicação de uma mensagem, ela ficará disponível para consumo, após o qual será descartada para liberar espaço. O desempenho do Kafka é efetivamente constante em relação ao tamanho dos dados, portanto, reter muitos dados não é um problema.

Na verdade, os únicos metadados retidos por consumidor são a posição do consumidor no log, chamada de "deslocamento". Esse deslocamento é controlado pelo consumidor: normalmente um consumidor avançará seu deslocamento linearmente à medida que lê as mensagens, mas na verdade a posição é controlada pelo consumidor e ele pode consumir as mensagens na ordem que desejar. Por exemplo, um consumidor pode redefinir para um deslocamento mais antigo para reprocessar.

Para encontrar o deslocamento inicial para ler no exemplo do consumidor simples do Kafka 0.8, eles dizem

O Kafka inclui duas constantes para ajudar, kafka.api.OffsetRequest.EarliestTime()encontra o início dos dados nos logs e começa a transmitir a partir daí, kafka.api.OffsetRequest.LatestTime()irá apenas transmitir novas mensagens.

Você também pode encontrar o código de exemplo lá para gerenciar o deslocamento na extremidade do 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 mencionei aqui, Purge Kafka Queue :

Testado no Kafka 0.8.2, para o exemplo de início rápido: Primeiro, adicione uma linha ao arquivo server.properties na pasta de configuração:

delete.topic.enable=true

então, você pode executar este comando:

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

Testado com 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: se você estiver excluindo pasta / s de tópico dentro de kafka-logs, mas não da pasta de dados do zookeeper, então verá que os tópicos ainda estão lá.

Como solução alternativa, você pode ajustar as configurações de retenção de tempo de execução por tópico, por exemplo, bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1( Retenção.bytes = 0 também pode funcionar)

Após um curto período, o kafka deve liberar o espaço. Não tenho certeza se isso tem alguma implicação em comparação com a recriação do tópico.

ps. Melhor trazer as configurações de retenção de volta, uma vez que o kafka feito a limpeza.

Você também pode usar retention.mspara persistir dados históricos

Abaixo estão scripts para esvaziar e excluir um tópico Kafka assumindo localhost como o servidor zookeeper e Kafka_Home está definido para o diretório de instalação:

O script a seguir esvaziará um tópico definindo seu tempo de retenção para 1 segundo e removendo a configuração:

#!/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 excluir totalmente os tópicos, você deve interromper qualquer (s) corretor (es) kafka aplicável e remover seu (s) diretório (s) do diretório de log kafka (padrão: / tmp / kafka-logs) e então executar este script para remover o tópico do zookeeper. Para verificar se ele foi excluído do zookeeper, a saída de ls / brokers / topics não deve mais incluir o tópico:

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

Tentamos basicamente o que as outras respostas estão descrevendo com nível moderado de sucesso. O que realmente funcionou para nós (Apache Kafka 0.8.1) é o comando class

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

Para usuários de cerveja

Se você está usando brewcomo eu e perdeu muito tempo procurando a kafka-logspasta infame , não tema mais. (e, por favor, deixe-me saber se isso funciona para você e várias versões diferentes do Homebrew, Kafka etc :))

Você provavelmente vai encontrar em:

Localização:

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

Como realmente encontrar esse caminho

(isso também é útil para basicamente todos os aplicativos que você instala por meio do brew)

1) brew services list

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

2) Abra e leia o que plistvocê encontrou acima

3) Encontre a linha definindo a server.propertieslocalização abri-la, no meu caso:

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

4) Procure a log.dirslinha:

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

5) Vá até aquele local e exclua os logs dos tópicos que deseja

6) Reinicie o Kafka com brew services restart kafka

Todos os dados sobre tópicos e suas partições são armazenados em tmp/kafka-logs/. Além disso, eles são armazenados em um formato topic-partionNumber, então se você quiser excluir um tópico newTopic, você pode:

• pare kafka
• apague os arquivos rm -rf /tmp/kafka-logs/newTopic-*

1. Pare o ZooKeeper e o Kafka
2. Em server.properties, altere o valor log.retention.hours. Você pode comentar log.retention.hourse adicionar log.retention.ms=1000. Isso manteria o registro do Tópico Kafka por apenas um segundo.
3. Inicie o zookeeper e o kafka.
4. Verifique no console do consumidor. Quando abri o console pela primeira vez, o registro estava lá. Mas quando abri o console novamente, o registro foi removido.
5. Mais tarde, você pode definir o valor de log.retention.hourspara o valor desejado.

A partir da versão 2.3.0 do kafka, há uma maneira alternativa de exclusão suave do Kafka (a abordagem antiga está obsoleta).

Atualize a retenção.ms para 1 s (1000 ms) e defina-a novamente após um minuto, para a configuração padrão, ou seja, 7 dias (168 horas, 604.800.000 em ms)

Exclusão 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'.

Configuração para o padrão: - 7 dias (168 horas, retenção.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

Ao excluir manualmente um tópico de um cluster kafka, você só pode verificar https://github.com/darrenfu/bigdata/issues/6 Uma etapa vital que muitas vezes se esquece na maioria das soluções é excluir o /config/topics/<topic_name>no ZK.

Eu 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

Eu uso o utilitário abaixo para limpar após a execução do meu teste de integração.

Ele usa a AdminZkClientAPI mais recente . A API mais antiga foi descontinuada.

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)

  }


}

Existe uma opção para excluir o tópico. Mas, isso marca o tópico para exclusão. O Zookeeper mais tarde exclui o tópico. Uma vez que isso pode ser imprevisivelmente longo, prefiro a abordagem Retenção