如何高效地将数据从Kafka移动到Impala表格？

Question

以下是当前流程的步骤：

1. Flafka 将日志写入HDFS上的“landing zone”。
2. 由Oozie调度的作业将完整的文件从landing zone复制到暂存区。
3. 使用暂存区作为其位置的Hive表对暂存数据进行“模式化”处理。
4. 将暂存表中的记录添加到永久Hive表中（例如，insert into permanent_table select * from staging_table）。
5. 通过在Impala中执行refresh permanent_table，可以在Hive表中获取数据。

我看着我建立的过程，感觉“有点不对劲”：有太多的中间步骤影响了数据的流动。

大约20个月前，我看到了一个演示，数据从Amazon Kinesis管道流式传输，并且可以由Impala近实时查询。我不认为他们做了什么太难看/复杂的事情。从Kafka流式传输数据到Impala是否有更有效的方法（可能是可以将数据序列化为Parquet的Kafka消费者）？

我想象“将数据流式传输到低延迟SQL”一定是一个相当普遍的用例，因此我有兴趣知道其他人是如何解决这个问题的。

- Alex Woolford

---

这里有一种备选架构，似乎跳数较少，可能相对更快。 - vmachan

这个有帮助吗？https://blog.cloudera.com/blog/2015/11/how-to-ingest-and-query-fast-data-with-impala-without-kudu/ - Dennis Jaheruddin

Answer 1

如果您需要将Kafka数据原样转储到HDFS，则最佳选项是使用Kafka Connect和Confluent HDFS连接器。您可以将数据转储到HDFS上的parquet文件中，然后在Impala中加载。您需要使用TimeBasedPartitioner分区器来使parquet文件每X毫秒生成一次（调整partition.duration.ms配置参数）。将以下内容添加到您的Kafka Connect配置中可能会起作用：

# 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

- Iñigo González

Answer 2

回答这个问题，如果是在2022年，我会说解决方案是从Kafka流式传输消息到Kudu，并将Impala与Kudu集成，因为它已经有了紧密的集成。

以下是Kudu的Impala模式示例：

CREATE EXTERNAL TABLE my_table
STORED AS KUDU
TBLPROPERTIES (
  'kudu.table_name' = 'my_kudu_table'
);

Apache Kudu支持SQL插入，并在底层使用自己的文件格式。或者，您可以使用Apache Phoenix，它支持插入和upserts（如果您需要确切的一次语义），并在底层使用HBase。

只要Impala是访问数据的最终方式，您就不必关心底层格式。

- aleksandertantus