hadoop - 将文件分区为日期从 kafka 写入 hdfs 的最有效方法是什么

Question

我正在研究应该通过 kafka 写入 hdfs 的项目。假设有在线服务器将消息写入kafka。每条消息都包含时间戳。 我想根据消息中的时间戳创建一个输出将是一个文件的作业。比如kafka中的数据是

 {"ts":"01-07-2013 15:25:35.994", "data": ...}
 ...    
 {"ts":"01-07-2013 16:25:35.994", "data": ...}
 ... 
 {"ts":"01-07-2013 17:25:35.994", "data": ...}

我想得到 3 个文件作为输出

  kafka_file_2013-07-01_15.json
  kafka_file_2013-07-01_16.json
  kafka_file_2013-07-01_17.json 

当然，如果我再次运行这个作业，并且队列中有一条新消息，例如

 {"ts":"01-07-2013 17:25:35.994", "data": ...}

它应该创建一个文件

  kafka_file_2013-07-01_17_2.json // second  chunk of hour 17

我看过一些开源，但其中大部分都是从 kafka 读取到某个 hdfs 文件夹。这个问题的最佳解决方案/设计/开源是什么

Answer 1

您绝对应该从 linkedIn 查看 Camus API 实现。 Camus 是 LinkedIn 的 Kafka->HDFS 管道。这是一个 mapreduce 作业，可以从 Kafka 加载分布式数据。看看这个 post我写了一个简单的例子，它从推特流中获取并根据推文时间戳写入 HDFS。

项目在 github 上可用 - https://github.com/linkedin/camus

Camus 需要两个主要组件来从 Kafka 读取和解码数据以及将数据写入 HDFS –

解码从 Kafka 读取的消息

Camus 有一组解码器，可以帮助解码来自 Kafka 的消息，解码器基本上扩展了 com.linkedin.camus.coders.MessageDecoder，它实现了根据时间戳对数据进行分区的逻辑。此目录中存在一组预定义的解码器，您可以基于这些编写自己的解码器。 camus/camus-kafka-coders/src/main/java/com/linkedin/camus/etl/kafka/coders/

向HDFS写入消息

Camus 需要一组 RecordWriterProvider 类，它扩展了 com.linkedin.camus.etl.RecordWriterProvider，它将告诉 Camus 什么是应该写入 HDFS 的有效载荷。一组预定义的 RecordWriterProvider 存在于这个目录，你可以根据这些编写你自己的。

camus-etl-kafka/src/main/java/com/linkedin/camus/etl/kafka/common