flume+kafka+Druid 流数据查询聚合工具

Druid是个用于海量查询、聚合的工具，项目比较新，资料方面还比较少。这里是一位博主fish_cool总结的资料，包括Druid的简介，适用的场景，用法等，很大部分也是从Druid官网上总结的，写得很不错。这篇文章要介绍的是实验室自己的系统设计思路。

整个系统的架构是 flume -> kafka -> druid，flume 与kafka的整合之前在搭建flume
-> kafka -> storm的时候有详细文档，通过 flume实现为kafka 的producer api，将程序打包成 jar放置在flume/lib 下，配置agent文件即可。

[align=justify]一、部署[/align]

[align=justify]部署druid需要先部署它的一些依赖，最主要的是两个： zookeeper和mysql ，zookeeper是 druid集群非常重要的一环，用于管理各个不同节点之间的工作和协调系统的运行。 mysql则是用来存储rules， segment等等的元数据。[/align]
[align=justify]zookeeper[/align]
[align=justify]1] 是druid官网上介绍部署 druid集群的文档，主要是zookeeper的部署，还有 druid几种重要节点配置文件的修改。[/align]
[align=justify]mysql[/align]
[align=justify]安装完mysql之后进入的 ${DRUID_HOME}/config/_common修改配置文件，图1所示参数是元数据存储修改为 mysql存储的示例。[/align]



[align=center]图 1 common.runtime.properties[/align]
[align=justify]- 类型type修改为 mysql[/align]
[align=justify]- 数据库的连接地址，注明主机名，端口号 (mysql的默认为3306 ）和数据库名（本例中为 druid，需要在mysql 中自己创建）[/align]
[align=justify]- 连接到数据库的用户名和密码[/align]
[align=justify]由于coodinator通过 zookeeper周期性的向mysql 查询segments的信息并决定卸载和载入 segments，因此zookeeper 和mysql都是必须的。[/align]
[align=justify]kafka[/align]
[align=justify]kafka 消息缓存作为一个 firehorse（druid 的数据源，输入源）， druid集成的很好。官网上给出的建议是如果在集群部署的话需要自己定制 kafka的comsumer api作为druid的输入源，否则在一致性方面可能会有问题，但是单点 druid和测试用直接用druid集成的 kafka配置就可以了，[2]给出了详细的运行步骤。在 druid要运行realtime 和historical必须要让它们知道你的数据源，数据的格式等信息，这些信息由 .spec为后缀的文件来指定，也是我们定制自己的数据源的重要文件，详细的各个字段在文档 [3]中dataSchema 都有提到。以 druid自带的examples/indexing/wikipedia.spec 文件来解释几个重要的字段。[/align]



[align=center]图 2 wikipedia.spec(上)[/align]

第4行指出的 dataSource指定数据源的名字，可以自定义，后面在做查询的时候你的 json文件需要提供一个dataSource数据源的字段，就是你所写的这个值。

图3是另外几个重要参数，包括将 zookeeper.connect修改为你自己的zookeeper的主机号和端口号， feed等。feed 这个字段非常重要，它就是指定去 kafka中哪个topic 中拉取数据的 topic名，必须跟flume 输出的topic， kafka作为缓存的topic 是同一个。另外一些参数是性能调优方面的，即箭头所指的持久化周期大小和 realtime服务的窗口大小，还有图 2，图3 没列举出来的可调优的字段，其他配置 dimensionsSpec和metricsSpec 等属性的都可以在文档 [3]中找到。



[align=center]图 3 wikipedia.spec(下)[/align]

curl

[align=left]部署druid完成之后，需要通过curl发送请求来测试或者查询结果。查询方式有两种，原理都一样，只是方式不一样，如图 4所示。[/align]

[align=center][/align]



[align=center]图 4 curl请求测试druid[/align]

-X 后面指定一种请求方式，比如图 4中的POST

-H 指定请求的Header 头信息

[align=left]-d 指定json 格式的内容或者一个 json文件( 用@引导 )[/align]

[align=left]二、使用步骤[/align]

1) flume -> kafka: kafkaSink程序导出jar包拷贝至flume/lib下；修改agent配置文件；创建topic

2) kafka -> druid: 修改${DRUID_HOME}/examples/indexing/wikipedia.spec文件，feed字段修改位kafka topic名，另外还有一些zookeeper地址之类的具体信息

3) 启动zookeeper服务

4) 启动kafka server服务

5) 启动flume agent服务

6) 启动druid realtime（配置文件要指向firehorse为kafka的文件）

7) 用flume avro-client发送数据

8) 用curl命令发送查询请求

reference

[1] http://druid.io/docs/latest/tutorials/tutorial-the-druid-cluster.html
[2] http://druid.io/docs/latest/tutorials/tutorial-a-first-look-at-druid.html
[align=justify][3] http://druid.io/docs/latest/ingestion/index.html[/align]