Speichert Apache Kafka die Nachrichten intern in HDFS oder einem anderen Dateisystem

Question

Wir haben eine Projektanforderung, die Daten bei Kafka Layer zu testen. So bewegen sich JSON-Dateien in den Hadoop-Bereich und kafka liest die Live-Daten in hadoop (Raw Json File). Nun muss ich testen, ob die vom anderen System gesendeten und von kafka gelesenen Daten gleich sein sollen.

Kann ich die Daten bei kafka validieren ?. Speichert Kafka die Nachrichten intern auf HDFS ?. Wenn ja, wird es in einer Dateistruktur gespeichert, die ähnlich der Struktur ist, die intern wie ein einzelner Ordner für einzelne Tabellen gespeichert wird.

Answer 1

Kafka speichert Daten in lokalen Dateien (dh lokales Dateisystem für jeden laufenden Broker). Für diese Dateien verwendet Kafka ein eigenes Speicherformat, das auf einer partitionierten Nur-Anhang-Protokollabstraktion basiert.

Das lokale Speicherverzeichnis kann über den Parameter log.dir konfiguriert werden. Diese Konfiguration geschieht individuell für jeden Broker, dh jeder Broker kann einen anderen Standort verwenden. Der Standardwert ist /tmp/kafka-logs.

Darüber hinaus hat jedes Thema mehrere Partitionen. Wie Partitionen verteilt werden, ist ein Kafka-internes Implementierungsdetail. Also solltest du dich jetzt darauf verlassen. Um den aktuellen Status Ihres Clusters zu erhalten, können Sie Metadaten zu Themen und Partitionen anfordern (siehe https://cwiki.apache.org/confluence/display/KAFKA/Finding+Topic+and+Partition+Leader für ein Codebeispiel). Bedenken Sie auch, dass Partitionen repliziert werden und wenn Sie schreiben, müssen Sie immer auf den Partitionsführer schreiben (wenn Sie einen KafkaProducer erstellen, wird automatisch der Leader für jede Partition gefunden, in die Sie schreiben).

Für weitere Informationen, nach https://cwiki.apache.org/confluence/display/KAFKA/Index

Answer 2

Ich glaube, Sie können, aber Sie müssen dies manuell tun. Sie können kafka jede Ausgabe in HDFS sinken lassen. Vielleicht ist meine Antwort ein bisschen spät und diese 'confluent' reference erschien danach, aber kurz kann man die folgenden tun:

• Angenommen, Sie haben alle Server ausgeführt werden (überprüfen Sie die konfluenten website)

• Ihr Anschluss erstellen:

  name=hdfs-sink 
  
  connector.class=io.confluent.connect.hdfs.HdfsSinkConnector 
  
  tasks.max=1 
  
  topics='your topic' 
  
  hdfs.url=hdfs://localhost:9000 
  
  flush.size=3 
  

• Hinweis: Der Ansatz geht davon aus, dass Sie ihre Plattform verwenden (confluent platform), die ich nicht verwenden.

• Den kafka-hdfs Streamer abfeuern.

Sie könnten auch weitere nützliche Details in diesem Stack Overflow discussion finden.