实时 OLAP 系统 Druid

0. Overview

后面将写几篇文章介绍一下 OLAP 的大数据系统架构。这里的 Druid 不是阿里巴巴的连接池,而是用于大数据实时 OLAP 的 Druid。

OLAP 和 OLTP 经常被拿到一起来讨论。其中 OLAP 的全称是 On-Line Analytical Processing,OLTP 的全称是 On-Line Transaction Processing。网上分析对比这两种系统的讨论很多都是长篇累牍,其实从系统角度来看 OLAP 和 OLTP 的最大区别无非是下面几点:

OLTP OLAP
查询数据
延迟 要求延迟低 可以容忍较高延迟
数据量
并发访问

OLTP 对应常见的关系型数据库,比如 MySQL 等。OLAP 又分实时 OLAP 和离线 OLAP。大数据的一些架构,比如常见 Hive + Hadoop,SparkSQL + HDFS,Kylin 等就是离线 OLAP,而一些监控告警系统这种对实时性要求比较高的系统就是实时 OLAP。

Druid 就属于实时 OLAP。我们是从去年的差不多这个时间开始使用 Druid 从 0 到 1 搭建了我们的实时 OLAP 系统,这套系统目前在线上运行半年,单个 DataSource 摄入的数据在百亿级别。在这个过程中遇到很多问题,也发现了 Druid 的一些局限性,这篇文章先简答介绍一下 Druid 的架构和一些特性,下一篇文章再介绍一些实践,可能不是最佳的。

1. 特性

Druid 很早就进入了 Apache 孵化器,但是现在还没有毕业。官网:https://druid.apache.org,Github: https://github.com/apache/incubator-druid

根据官方文档,Druid 的核心特性主要包括:

  • 列式存储。列式存储的优势在于查询的时候可以只返回指定的列的数据,其次同一列数据往往具有很多共性,这带来另一个好处就是存储的时候压缩效果比较好。
  • 可扩展的分布式架构。
  • 并行计算。
  • 数据摄入支持实时和批量。这里的实时的意思是输入摄入即可查。如果大家看过我之前关于实时计算的文章,应该猜到了这就是典型的 lambda 架构,后面再细说。
  • 运维友好。
  • 云原生架构,高容错性。
  • 支持索引,便于快速查询。
  • 基于时间的分区
  • 自动聚合。

不知道官方是不是为了刻意凑数,正好十条。其中很多特性其实应该算是 OLAP 系统的共同特性,比如列式存储等。当时我选型使用 Druid 的时候,其实最吸引我的主要是下面三条:

  • 实时摄取可查询。换句话说就是数据查询无延迟,这个在一些对实时性要求比较高的场景下,比如监控告警,还是很重要的。
  • 自动实时聚合。
  • 高效的索引结构便于查询。

2. 架构

Druid 的架构在我看来还是比较复杂的,包含 6 个不同的组件。

  • Coordinator:顾名思义,Coordinator 就是协调器,主要负责 segment 的分发等。比如我们只保存 30 天的数据,这个规则就是由 Coordinator 来定时执行的。

  • Overlord:处理数据摄入的 task,将 task 提交到 MiddleManager。比如使用 Tranquility 做数据摄入的时候,每个 segment 都会生成一个对应的 task。

  • Broker : 处理外部请求,并对结果进行汇总。

  • Router : Router 相当于多个 Broker 前面的路由,不是必须的。

  • Historical :Historical 可以理解为将 segment 存储到本地,相当于 cache。相比于 Deep Storage 的,Historical 将 segment 直接存储到本地磁盘,只有 segment 存储到本地才能被查询。其实这个地方是有点异于直观感受的。正常我们可能会认为查询先查本地,如果本地没有数据才去查 Deep Storage,但是实际上如果本地没有相应的 segment,则查询是无法查询的。

    Historical 处理那些 segment 是由 Coordinator 指定的,但是 Historical 并不会和 Coordinator 直接交互,而是通过 Zookeeper 来解耦。

  • MiddleManager : MiddleManager 可以认为是一个任务调度进程,主要用来处理 Overload 提交过来的 task。每个 task 会以一个 JVM 进程启动。

各个组件之间的交互如下:

druid-architecture.png

根据线条,上图主要关注三个部分:

  • Queries: Routers 将请求路由到 Broker,Broker 向 MiddleManager 和 Historical 进行数据查询。这里 MiddleManager 主要负责查询正在进行摄入的数据查询,比如现在正在摄入 12:00 ~ 13:00 的数据,那么我们要查询就去查询 MiddleManager,MiddleManager 再将请求分发到具体的 peon,也就是 task 的运行实体上。而历史数据的查询是通过 Historical 查询的,然后数据返回到 Broker 进行汇总。这里需要注意的时候数据查询并不会落到 Deep Storage 上去,也就是查询的数据一定是 cache 到本地磁盘的。很多人一个直观理解查询的过程应该是先查询 Historical,Historical 没有这部分数据则去查 Deep Storage。Druid 并不是这么设计的。

  • Data/Segment: 这里包括两个部分,MiddleManager 的 task 在结束的时候会将数据写入到 Deep Storage,这个过程一般称作 Segment Handoff。然后 Historical 定期的去下载 Deep Storage 中的 segment 数据到本地。

  • Metadata: Druid 的元数据主要存储到两个部分,一个是 Metadata Storage,这个一般是 MySQL 等关系型数据库;另一个是 Zookeeper。下图是 Druid 在 Zookeeper 中的 znode。zk 的作用主要是用来给各个组件进行解耦。

    zk.png

3. 数据存储

Druid 的数据存储单位是 segment,segment 按时间粒度(可以通过参数 segmentGranularity 指定)划分。每个 segment 会被存储到 Deep Storage 和 Historical 进程所在的节点上,当然 segment 可以是有多个备份的,这样查询的时候就可以实现并行查询,并不是为了高可用,高可用通过 Deep Storage 保证。

Druid 的数据格式如下:


druid-column-types.png

分成三个部分:

  • Timestamp:时间戳信息
  • Dimension:维度信息
  • Metrics: 一般是数值型

Druid 会自动对数据进行 Rollup,也就是聚合。如果时间粒度是一小时,那么在这一个小时内维度相同的数据会被合并为一条,Timestamp 都变成整点,metrics 会根据聚合函数进行聚合,比如 sum, max, min 等,注意是没有平均 avg 的。Timestamp 和 Metrics 直接压缩存储即可,比较简单。下面重点说一下维度的存储。

Druid 的一大亮点就是支持多维度实时聚合查询,简单来说就是 filter 和 group。而实现这个特性的关键技术主要两点:bitmap + 倒排。

首先,Druid 会将维度值编码映射成数字 ID,类似数据仓库中的维度表,主要是为了存储节省空间。比如上面图中的 Page 维度:Justin Bieber 被编码成 0,Ke$ha 被编码成 1。对于 Username 维度:Boxer -> 0,Reach -> 1,Helz -> 0,Xeno -> 1。

然后 Page 这列数据就会被存储为 [0,0,1,1]。

最后是位图,用来表示对于某个维度的某个值,有哪些列包含了这个值,比如:

  • Justin Bieber: [1,1,0,0]
  • Boxer: [1,0,0,0]

那么 filter 查询 Page='Justin Bieber' and Username='Boxer',直接将 1100 和 1000 做位运算 and 即可。group 也是类似。

上面的位图,其实也是一种倒排,常规的倒排后面的 list 中直接包含的是 Document ID,这里直接表示成位图,其实是异曲同工。

4. 数据摄入

前面简单提到 Druid 的数据摄入支持实时流模式和批模式,也就是典型的 Lambda 架构。Lambda 架构简单来说就两点:

  • 通过实时处理保证实时性
  • 通过批处理保证数据完整性和准确性

如果看过我之前的关于 Google DataFlow 的文章,当时作者就大肆批评了 lambda 架构,然后在 Google 内部是通过 MillWheel 支持 exactly-once 语义来避免了 lambda 架构。在 druid 中的数据摄入官方支持了多种方式,关于各种方式的对比可以用如下一个图来概括。

ingest.png

关于上图中的 ”Can handle late data“ 做一下简单说明,我们上面在数据存储一节有说到 Druid 的底层存储使用了 segment 结构。举个例子,如果时间粒度是 1 个小时,那么 12:00 ~ 13:00 的数据就会存储到一个 segment 里面。但是这里有一个小问题需要考虑一下,就是这个 segment 的数据什么时候 ready 我怎么知道呢?这个在流处理中一种常规的做法是 watermark,简单来说就设置一个可以接受的时间延迟,比如 5 分钟,那么 12:00 ~ 13:00 会一直接受数据直到 13:05,然后之后这个 segment 就会被 handoff 掉,12:00 ~ 13:00 之间的数据就不再接受了。这个过程就叫做 ”handle late data“。然后我们发现上图中 Tranquility 是不支持 late data 处理的,这个是需要特别注意的。

从上图我们可以看到 Native batch 和 Hadoop 都对应了 Lambda 架构中的批处理,而 Tranquility 则对应了 Lambda 架构中实时处理,是一种 push 的方式。然后这里还有一种方式叫 Kafka Indexing Service,这种方式通过 pull 的方式来摄取数据,我们也可以看到通过 Kafka Indexing Service 这一种服务其实就可以完成数据摄取并满足所有需求,不然就要通过两种方式联合使用。但是使用 Kafka Indexing Service 的最大问题就是和 Kafka 强耦合。

因为我们的业务是在阿里云公有云上,然后所有数据采集都使用了阿里云的日志服务(SLS)来处理的,所以这里我们并不能使用 Kafka index。这里我们使用的方式是 Tranquility + Hadoop 的方式来进行数据摄取。单个 DataSource 的数据摄入量达到百亿级别。

5. 查询

5.1 Natvie

Druid 最开始的时候是不支持 SQL 查询的,原生查询是通过查询 Broker 提供的 http server 来实现的,如下:

curl -X POST '<queryable_host>:<port>/druid/v2/?pretty' -H 'Content-Type:application/json' -H 'Accept:application/json' -d @<query_json_file>

下面是一个简单的 json 查询示例。

{
  "queryType": "timeseries",
  "dataSource": "sample_datasource",
  "intervals": [ "2012-01-01T00:00:00.000/2012-01-03T00:00:00.000" ],
  "granularity": "day",
  "aggregations": [
    { "type": "longSum", "name": "sample_name1", "fieldName": "sample_fieldName1" },
    { "type": "doubleSum", "name": "sample_name2", "fieldName": "sample_fieldName2" }
  ],
  "context": {
    "grandTotal": true
  }
}

同时社区也提供了很多种语言的 client 用来做 Druid 的查询,比如我们使用的 Java 的 client zapr/druidry ,关于更多语言的 client,可以参考这里 client libraries

Druid 的查询类型有下面几种:

  • 聚合查询(Aggregation Queries):

    • Timeseries : 可以简单理解为性能更好的 select。
    • TopN : TopN 相当于 GroupBy 加 Ordering,相同的查询我们正常也可以通过 GroupBy 查询来实现,但是 TopN 的性能更好。TopN 的底层实现也是比较直观的,将并行查询的每个查询的结果的 TopK 结果返回给 Broker,由 Broker 进行聚合汇总。注意这里返回的结果是 K 条记录,而不是 N 条记录,K 默认值为 max(1000, threshold) 决定(threshold 由用户指定,就相当于 TopN 中的 N)。
    • GroupBy : GroupBy。
  • 元数据查询(Metadata Queries)

    Druid 的元数据一般是存储到 MySQL 中,包含一些 dataSource,segment 的元信息。

    metadata.png

Druid 提供的元数据查询有下面三种

  • Time Boundary: 用来查询查询指定模式的数据第一次出现和最近一次出现的时间。

  • Segment Metadata:返回 segment 的元信息,包括维度信息等。

  • Datasource Metadata:返回 dataSource 的元信息。

  • 搜索查询(Search Queries)

  • 范围查询 (Scan):scan 的结果是以流模式返回的,也就是 client 真正读取的时候才会占用内存。

  • Select: 官方已经不建议使用 Select 查询。这里就不在介绍了。

Druid 的底层存储由于是使用时间来做分片的,所以查询的时候一定需要带上时间区间。

我在上面说过一次 Druid 的 Rollup 不支持 average,也就是平均值,那么如果我查询的时候要查询平均值应该怎么做呢?(其实查询平均值是一个非常常见的需求,关于为了 Druid 的 Rollup 不支持 average,欢迎留言讨论。)

答案是 postaggregate,druid 在查询的时候可以定义聚合操作,是查询的时候直接计算的。同时 druid 还提供了针对聚合后的值的聚合操作,叫做 postaggregate。一个简单的查询 json 文件示例。

{
  "queryType": "timeseries",
  "dataSource": "sample_datasource",
  "granularity": "day",
  "descending": "true",
  "filter": {
    "type": "and",
    "fields": [
      { "type": "selector", "dimension": "sample_dimension1", "value": "sample_value1" },
      { "type": "or",
        "fields": [
          { "type": "selector", "dimension": "sample_dimension2", "value": "sample_value2" },
          { "type": "selector", "dimension": "sample_dimension3", "value": "sample_value3" }
        ]
      }
    ]
  },
  "aggregations": [
    { "type": "longSum", "name": "sample_name1", "fieldName": "sample_fieldName1" },
    { "type": "doubleSum", "name": "sample_name2", "fieldName": "sample_fieldName2" }
  ],
  "postAggregations": [
    { "type": "arithmetic",
      "name": "sample_divide",
      "fn": "/",
      "fields": [
        { "type": "fieldAccess", "name": "postAgg__sample_name1", "fieldName": "sample_name1" },
        { "type": "fieldAccess", "name": "postAgg__sample_name2", "fieldName": "sample_name2" }
      ]
    }
  ],
  "intervals": [ "2012-01-01T00:00:00.000/2012-01-03T00:00:00.000" ]
}

5.2 SQL

SQL 在大数据系统,尤其是 OLAP 中的重要性是不言而喻的。所以早期看到 Druid 不支持 SQL 查询,我是非常诧异的,后面果不其然,Druid 还是推出了 SQL 查询。这一层构建与 Native 请求之上,也就是说 SQL 会被解释成 Native 的查询,然后去请求 Broker。

Druid SQL 解析基于 Apache Calcite,说起 Apache Calcite 是一个业界使用非常广泛的 SQL 语法解析模块,如果没有记错, Hive 使用的好像也是它。

Druid SQL 值得一提的是提供了非常多的 function,包括数值计算,字符串操作,时间操作等。举个例子,其中一个字符串操作函数叫做 REGEXP_EXTRACT(expr, pattern, [index]) 对 expr 做正则匹配,并提取特定的字段。使用这个函数可以做非常多的事情。但是 function 有的时候对于 SQL 的执行计划优化并不是非常友好,不知道这里 Druid 团队是如何权衡的。

6. 其他

其他有一些值得讨论的话题列在这里。

6.1 明细查询

由于 Druid 会对存储的数据做 Rollup,正常情况下是不能存储明细的。但是如果是你一定需要明细的话,有个办法就是将所有所有的列,包括 metric,都设置成 dimension,同时将聚合粒度设置到可以接受的粒度,比如秒。

6.2 高基数

这里的高基数指的是 Druid 的 Dimension 的值可能会有非常多的值,这样引入一个问题就是存储的时候会消耗比较大的空间,同时对于 CPU 的占用也会有一定程度的影响。

7. 总结

整体来看,Druid 算是一个优秀的实时 OLAP 系统,虽然有一些地方设计的并不是尽善尽美,但是瑕不掩瑜。这篇文章简单介绍一些 Druid 的整体情况,希望可以给使用 Druid 的同学做一些参考。下一篇文章将会介绍一下我们过去一年基于 Druid 的实践情况以及一些踩过的坑。

最后有一个问题我非常好奇,横向对比 Apache 基金会的项目,Druid 在很多方面都是可圈可点的,但是为什么现在还没有从 Apache 毕业,实在是令人困惑。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,456评论 5 477
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,370评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,337评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,583评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,596评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,572评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,936评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,595评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,850评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,601评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,685评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,371评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,951评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,934评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,167评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 43,636评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,411评论 2 342

推荐阅读更多精彩内容

  • Druid.io(以下简称Druid)是面向海量数据的、用于实时查询与分析的OLAP存储系统。Druid的四大关键...
    大诗兄_zl阅读 6,442评论 0 9
  • Druid io总体设计 1.Druid模块架构 1.1 Druid简介 最新版本的Druid采用了位图索引、字典...
    小武大讲堂阅读 1,814评论 0 2
  • 我们知道Druid能够同时提供对大数据集的实时摄入和高效复杂查询的性能,主要原因就是它独到的架构设计和基于Data...
    零度沸腾_yjz阅读 21,499评论 3 17
  • #refer1:http://www.cnblogs.com/xd502djj/p/6408979.html#re...
    liuzx32阅读 1,898评论 0 1
  • “结婚是什么感受? 就是把你手机里所有的音乐都删掉, 只留下最喜欢的一首, 然后无限循环, 由喜欢到厌倦再到习惯。...
    柒月同学阅读 421评论 0 1