jak skutecznie przenieść dane z Kafki do tabeli Impala?

Question

Oto kroki do obecnego procesu:

1. Flafka zapisuje logi do „strefy lądowania” na HDFS.
2. Zlecenie, zaplanowane przez Oozie, kopiuje kompletne pliki ze strefy lądowania do strefy postoju.
3. Dane dotyczące przemieszczania są "schematem" według tabeli gałęzi, która wykorzystuje obszar przemieszczania jako lokalizację.
4. Zapisy z tabeli pomostowej są dodawane do stałej tabeli Ula (np. insert into permanent_table select * from staging_table).
5. Dane z tabeli Hive są dostępne w Impala, wykonując refresh permanent_table w Impala.

patrzę w procesie I został zbudowany i „pachnie” zła: jest zbyt wiele pośrednich etapów, które upośledzają przepływ danych.

Około 20 miesięcy temu widziałem demo, w którym przesyłane były dane z rury Amazon Kinesis, i było możliwe do przeprowadzenia w niemal rzeczywistym czasie przez Impalę. Nie sądzę, żeby zrobili coś tak brzydkiego/zawiłego. Czy istnieje skuteczniejszy sposób przesyłania danych z Kafki do Impala (być może konsumenta Kafki, który może serializować do Parkietu)?

Wyobrażam sobie, że "przesyłanie strumieniowe danych do SQL o niskim opóźnieniu" musi być dość powszechnym przypadkiem użycia, dlatego chciałbym wiedzieć, jak inni rozwiązali ten problem.

Answer 1

Jeśli chcesz zrzucić swoje dane Kafki tak, jak jest to do HDFS, najlepszą opcją jest użycie złącza Kafka Connect i Confluent HDFS.

Możesz zrzucić dane do pliku parket na HDFS, który możesz załadować w Impala. Będziesz potrzebować Myślę, że będziesz chciał użyć partycji TimeBasedPartitioner do tworzenia plików parkietu co każde 10 milisekund (dostrajanie parametru konfiguracyjnego partition.duration.ms).

Addign coś takiego do konfiguracji Kafka Połącz może załatwić sprawę:

# 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