come spostare in modo efficiente i dati da Kafka a un tavolo Impala?

Question

Ecco i passi per il processo in corso:

1. Flafka scrive registri in un 'zona di atterraggio' su HDFS.
2. Un lavoro, pianificato da Oozie, copia i file completi dalla zona di atterraggio a un'area di gestione temporanea.
3. I dati di staging sono 'schema-ified' da una tabella Hive che utilizza l'area di staging come posizione.
4. I record dalla tabella di staging vengono aggiunti a una tabella Hive permanente (ad esempio insert into permanent_table select * from staging_table).
5. I dati, dalla tabella Hive, sono disponibili in Impala eseguendo refresh permanent_table in Impala.

guardo il processo ho costruito e "odori" cattivo: ci sono troppi passaggi intermedi che compromettono il flusso di dati.

Circa 20 mesi fa, ho visto una demo in cui i dati venivano trasmessi da una pipe Amazon Kinesis ed era interrogabile, quasi in tempo reale, da Impala. Non credo che abbiano fatto qualcosa di così brutto/contorto. Esiste un modo più efficiente di trasmettere i dati da Kafka a Impala (probabilmente un consumatore di Kafka che può serializzare su Parquet)?

Immagino che "lo streaming di dati su SQL a bassa latenza" debba essere un caso d'uso abbastanza comune, quindi sono interessato a sapere in che modo altre persone hanno risolto questo problema.

Answer 1

Se è necessario scaricare i dati di Kafka come in HDFS, l'opzione migliore è utilizzare il connettore HDFS di Kafka Connect e Confluent.

È possibile scaricare i dati in un file parket su HDFS che è possibile caricare in Impala. Avrai bisogno di pensare che vorrai usare un partizionatore TimeBasedPartitioner per creare file parquet ogni X millisecondi (sintonizzando il parametro di configurazione partition.duration.ms).

Addign qualcosa di simile per la configurazione di Kafka Collegare potrebbe fare il trucco:

# 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