Druid如果需要通过Kafka加载实时数据并进行OLAP，需要下载Druid的另一个组建，也就是tranquility-distribution，下载地址：http://static.druid.io/tranquility/releases/tranquility-distribution-0.8.2.tgz

    tranquility-distribution提供了Server和Kafka两种通过流来加载数据方式，由于业务环境的因素，对于Server的方式没有做什么研究，感兴趣的伙伴可以前往Druid的官网进行围观。接下来，详细说一下我研究的Kafka加载数据方式。

1.  Druid的数据组成

    无论使用何种方式加载自己的数据，都需要明白Druid的数据组成，因为需要通过配置文件描述加载的数据如进行存储，存储的格式是什么样子的。

    在Druid中数据的存储为三个部分，分别为：timestampSpec、dimensionsSpec和metricsSpec，其中timestampSpec是时间戳字段，这个字段是进行Query时的主要字段，这也是Druid可以作为时序数据库使用的主要因素。dimensionsSpec是维度字段，可以理解为分组条件或MapReduce中的Key。metricsSpec为根据维度对数据进行加工进而产生的数据，比如：count、sum、max、min、first、last等等。

    举个栗子

    原始数据是这个样子的

{"unit": "milliseconds", "http_method": "GET", "value": 70, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www2.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 116, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 35, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/80", "metricType": "request/latency", "server": "www1.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 44, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 40, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www1.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 84, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/33", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 42, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/54", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 115, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 32, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 82, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/15", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 123, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/22", "metricType": "request/latency", "server": "www3.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 57, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www3.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 26, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www1.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 44, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/45", "metricType": "request/latency", "server": "www3.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 132, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/42", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 113, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/70", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 144, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 41, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www1.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 90, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/94", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 89, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 114, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www3.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 29, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www5.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 25, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/get/41", "metricType": "request/latency", "server": "www4.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 88, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/", "metricType": "request/latency", "server": "www2.example.com"} {"unit": "milliseconds", "http_method": "GET", "value": 137, "timestamp": "2017-11-28T03:16:20Z", "http_code": "200", "page": "/list", "metricType": "request/latency", "server": "www4.example.com"} 

我们对这些数据进行加工，其中timestampSpec为timestamp，dimensionsSpec包括page、server，metricsSpec为count和value的sum，加工之后Druid中存储的数据格式将变成下面的样子

{"timestamp":"2017-11-28T03:16:20Z", "page":"/list", "server":"www2.example.com", "count":1, "valueSum":70} {"timestamp":"2017-11-28T03:16:20Z", "page":"/list", "server":"www4.example.com", "count":2, "valueSum":253} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/80", "server":"www1.example.com", "count":1, "valueSum":35} {"timestamp":"2017-11-28T03:16:20Z", "page":"/", "server":"www5.example.com", "count":4, "valueSum":320} {"timestamp":"2017-11-28T03:16:20Z", "page":"/list", "server":"www1.example.com", "count":2, "valueSum":66} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/33", "server":"www4.example.com", "count":1, "valueSum":84} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/54", "server":"www5.example.com", "count":1, "valueSum":42} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/15", "server":"www4.example.com", "count":1, "valueSum":82} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/22", "server":"www3.example.com", "count":1, "valueSum":123} {"timestamp":"2017-11-28T03:16:20Z", "page":"/", "server":"www3.example.com", "count":2, "valueSum":171} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/45", "server":"www3.example.com", "count":1, "valueSum":44} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/42", "server":"www5.example.com", "count":1, "valueSum":132} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/70", "server":"www4.example.com", "count":1, "valueSum":113} {"timestamp":"2017-11-28T03:16:20Z", "page":"/list", "server":"www5.example.com", "count":1, "valueSum":144} {"timestamp":"2017-11-28T03:16:20Z", "page":"/", "server":"www1.example.com", "count":1, "valueSum":41} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/94", "server":"www5.example.com", "count":1, "valueSum":90} {"timestamp":"2017-11-28T03:16:20Z", "page":"/", "server":"www4.example.com", "count":1, "valueSum":89} {"timestamp":"2017-11-28T03:16:20Z", "page":"/get/41", "server":"www4.example.com", "count":1, "valueSum":25} {"timestamp":"2017-11-28T03:16:20Z", "page":"/", "server":"www2.example.com", "count":1, "valueSum":88}

2.通过Kafka加载数据，并完成上面例子的加工操作

    在导入数据的过程中，我们需要对获取的数据如何加工进行配置，具体kafka.json配置如下：

{   "dataSources" : {     "test1" : { --数据源名称，与下方的dataSource的值要完全相同       "spec" : { --开始描述         "dataSchema" : { --开始数据表描述           "dataSource" : "test1", --数据源名称，注意上下对应           "parser" : {             "type" : "string",             "parseSpec" : {               "timestampSpec" : {                 "column" : "timestamp",                 "format" : "auto"               },               "dimensionsSpec" : { --维度字段都有那些                 "dimensions" : ["page", "server"]               },               "format" : "json" --格式化工具类型，对应从kafka中加载的数据的格式             }           },           "granularitySpec" : { --合并、分卷设置             "type" : "uniform",             "segmentGranularity" : "hour", --按照小时进行分卷设置             "queryGranularity" : "none" --默认合并模式，采用按秒进行合并的方式进行           },           "metricsSpec" : [ --合并计算列             {               "type" : "count", --count操作               "name" : "count" --列名             },             {               "type" : "longSum", --对long型数据的sum操作，当然还有doubleSum               "name" : "value_sum", --列名               "fieldName" : "value" --需要进行Sum的属性名             }           ]         },         "ioConfig" : {           "type" : "realtime" --实时加载         },         "tuningConfig" : {           "type" : "realtime",           "maxRowsInMemory" : "100000",  --内存中最大保存行号           "intermediatePersistPeriod" : "PT10M", --内存提交到磁盘的时间间隔           "windowPeriod" : "PT10M" --时间窗口缓冲时间         }       },       "properties" : {         "task.partitions" : "1",         "task.replicants" : "1",         "topicPattern" : "sunz"       }     }   },   "properties" : {     "zookeeper.connect" : "localhost",  --Druid使用的Zookeeper集群所在主机     "druid.discovery.curator.path" : "/druid/discovery",     "druid.selectors.indexing.serviceName" : "druid/overlord",     "commit.periodMillis" : "15000",     "consumer.numThreads" : "2",     "kafka.zookeeper.connect" : "devhadoop241", --kafka所使用的Zookeeper所在主机     "kafka.group.id" : "tranquility-kafka" --Topic名称   } } 

    启动方式比较简单，只需要执行

bin/tranquility kafka -configFile conf/kafka.json

    启动tranquility之后，就可以通过kafka加载数据了，如何建立Kafka生产者这个就不再赘述。

3.关于分段、聚合、内存提交及时间窗口四个时间相关设置的详解

    在上面的配置中，有四个属性的配置是至关重要的，分别为segmentGranularity--数据分段/分卷时间，queryGranularity数据碰撞时间、intermediatePersistPeriod内存持续时间/提交硬盘持久化时间间隔，windowPeriod窗口期/过期数据延迟加载周期。

    segmentGranularity

    Druid的数据是分卷储存的，通过这个属性，我们可以设置具体的分卷周期，比如：按分钟、小时、天、周、月、年进行分卷，具体分卷方式可以根据具体业务情况和数据规模进行设置。

    queryGranularity

    数据碰撞周期，这个周期所设置的是在何种时间颗粒度上对数据进行聚合操作，默认设置为none，如果设置为none则根据数据源中所设置时间戳字段进行聚合，如果时间戳一样则自动聚合，当然也可以设置为分钟和小时等，当然这个时间的设置一定要小于数据分卷周期

    intermediatePersistPeriod

    加载的数据在内存中驻守时间长度，加载的数据由于聚合需要，会先放置在内存中，一段时间之后统一提交到磁盘进行持久化操作，这个属性是影响性能的重要属性，设置的周期过大，内存消耗严重，导致假死，设置的过小，导致每次聚合都将操作磁盘，效率低下

    windowPeriod

    时间窗口期的作用是对数据延迟加载的设置，比如我们的数据分卷方式是按照小时进行分卷的，那么在只能realtime模式下，只能处理当前时段的数据，也就是意味着比如现在是12：04，那么只能处理12:00--13:00之间产生的数据，而由于11:00--12:00访问量过大，导致上一时段的数据还没有处理完，这部分数据将被抛弃，为了防止此类的数据丢失，我们设置时间窗口，对处理数据的时段设置进行缓冲，比如上一配置中，设置的是10分钟的时间窗口，也就意味着，在12:10之前，依然可以处理上一时段的数据。

    四属性的设置策略

    从上面的配置的作用可以看出，以上四属性的配置，要遵循 segmentGranularity > queryGranularity >= intermediatePersistPeriod >= windowPeriod的方式来进行配置才能保证处理性能。当然也需要衡量查询需求、数量吞吐量和硬件资源。