Wie kann man Daten effizient von Kafka in eine Impala-Tabelle verschieben?

Question

Hier sind die Schritte, um den aktuellen Prozess:

1. Flafka schreibt Protokolle in einer ‚Landezone‘ auf HDFS.
2. Ein von Oozie geplanter Job kopiert komplette Dateien von der Landezone in einen Bereitstellungsbereich.
3. Die Staging-Daten werden von einer Hive-Tabelle "schemaisiert", die den Staging-Bereich als Speicherort verwendet.
4. Datensätze aus der Staging-Tabelle werden einer permanenten Hive-Tabelle hinzugefügt (z. B. insert into permanent_table select * from staging_table).
5. Die Daten aus der Hive-Tabelle sind in Impala verfügbar, indem refresh permanent_table in Impala ausgeführt wird.

ich auf den Prozess sehe ich gebaut habe und es „riecht“ schlecht: es gibt zu viele Zwischenschritte, die den Datenfluss beeinträchtigen.

Vor etwa 20 Monaten habe ich eine Demo gesehen, bei der Daten von einer Amazon Kinesis-Pipe gestreamt wurden und fast in Echtzeit von Impala abgefragt werden konnten. Ich nehme nicht an, dass sie etwas so Hässliches/Verbogenes getan haben. Gibt es eine effizientere Möglichkeit, Daten von Kafka zu Impala zu übertragen (möglicherweise ein Kafka-Konsument, der zu Parquet serialisieren kann)?

Ich stelle mir vor, dass "Streaming-Daten zu Low-Latency-SQL" ein ziemlich häufiger Anwendungsfall sein muss, und ich bin interessiert zu wissen, wie andere Leute dieses Problem gelöst haben.

Answer 1

Wenn Sie Ihre Kafka-Daten unverändert in HDFS speichern möchten, verwenden Sie am besten Kafka Connect und Confluent HDFS Connector.

Sie können die Daten entweder in eine Parket-Datei auf HDFS speichern, die Sie in Impala laden können. Sie werden brauchen Ich denke, dass Sie einen TimeBasedPartitioner-Partitionierer verwenden sollten, um Parkett-Dateien alle X Millisekunden zu machen (Optimierung des Konfigurationspartition partition.duration.ms).

Addign so etwas zu Ihrer Kafka Connect-Konfiguration könnte den Trick tun:

# Don't flush less than 1000 messages to HDFS 
flush.size = 1000 

# Dump to parquet files 

format.class=io.confluent.connect.hdfs.parquet.ParquetFormat 

partitioner.class = TimebasedPartitioner 

# One file every hour. If you change this, remember to change the filename format to reflect this change 
partition.duration.ms = 3600000 
# Filename format 
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH/'minute'=mm