大数据分析-常见分布式OLAP查询引擎研究

大数据查询分析-分布式OLAP查询引擎设计

Bigdata OLAP Product= Cube + OLAP(QueryParser+QueryOptimizer+QueryEngine) + FileSystem(HDFS/GFS/S3)

• OLAP类型:ROLAP/BigQuery-OLAP/RT-OLAP/MOLAP,其中Kylin是一种针对大数据场景设计的特殊MOLAP
• 当前OLAP技术领域有大数据量分析需求,不包含查询引擎与数据存储优化的轻量级方案(Mondrian)应用场景受限
• 越来越多关注查询本身而不是过多关注Cube/Dimension/Measure等传统数据仓库所遵循的通用标准,这样才能更加适应当前大数据领域的现实需求
• Tableau作为优秀可视化分析工具对大数据量分析能力有所欠缺

OLAP场景的关键特征

• 大多数是读请求
• 数据总是以相当大的批次(> 1000 rows)进行写入
• 不修改已添加的数据
• 每次查询都从数据库中读取大量的行，但是同时又仅需要少量的列 宽表，即每个表包含着大量的列
• 较少的查询(通常每台服务器每秒数百个查询或更少)
• 对于简单查询，允许延迟大约50毫秒
• 列中的数据相对较小: 数字和短字符串(例如，每个URL 60个字节) 处理单个查询时需要高吞吐量(每个服务器每秒高达数十亿行)
• 事务不是必须的
• 对数据一致性要求低
• 每一个查询除了一个大表外都很小 查询结果明显小于源数据，换句话说，数据被过滤或聚合后能够被盛放在单台服务器的内存中

1.ROLAP引擎 - Mondrian

特点:多维数据建模+无内置查询引擎

[Mondriad-ROLAP分析]

Comments:敏捷BI多是基于Mondrain框架搭建的OLAP查询引擎

2.SQLonHadoop查询引擎 - (Impala/Presto/Dremel/Redshift)

特点:DistrubutedSQLQueryEngine分布交互式查询

• [SQLonHadoop技术分析]
• [Impala]
• [PrestoDB]
• BigQuery&Dremel
• Amazon Redshift

3.关于Kylin-MOLAP

特点:Cube预处理+多维数据查询(QueryEngine应弱于Impala/Presto)

3.1.Kylin架构

• DataSource:Hive
• MapReduce聚合计算
• Spark内存计算
• AggregateTable:HBase
• pre-aggregation预聚合
• 增量CubeSegment/CubeSegmentMerge
• Trie树维度值编码

3.2.TechnicalConcepts

• Star Schema/Snowflake Schema
• Cube
• DIMENSION & MEASURE
• CUBE ACTIONS

3.3 表描述

• FactTable
• LookupTable(事实描述表)
• DimensionTable

3.4 Measure计算

• Sum
• Count
• Max
• Min
• Average
• Distinct Count(based on HyperLogLog近似值估值)

4.RT(Realtime类搜索)OLAP查询引擎 - Druid/Pinot

Druid是基于MOLAP模型的空间换时间方案。优点在于查询性能的整体提升,缺点在于数据多维分析的局限性

4.1.Druid特点

特点:类搜索引擎+增量计算+数据实时写入

• 整个系统同时提供了对离线数据分析和在线实时数据分析的支持

• 可插拔的查询系统,支持多种存储系统

  • 亚秒响应的交互式查询。支持较高并发，为面向用户的平台提供Olap查询(注意这是相比其他OLAP的强大优势)。
  • 支持实时导入,导入即可被查询。支持高并发导入。
  • 采用分布式shared-nothing的架构，可以扩展到PB级。
  • 支持聚合函数,count和sum,以及使用javascript实现自定义UDF。
  • 支持复杂的Aggregator，近似查询的Aggregator例如HyperLoglog以及Yahoo开源的DataSketches。
  • 支持Groupby,Select,Search查询。(Groupby性能较差,推荐timeseries/TopN)
  • 不支持大表之间的Join，但其lookup功能满足和维度表的Join
  • 列存储,倒排索引,RollUP(汇总/上卷),roaring或conciseBitmap位图索引+LZ4数据压缩

4.2.Druid架构分析

Druid ingestion specs define this granularity as the queryGranularity of the data. The lowest supported queryGranularity is millisecond.
Druid shards are called segments and Druid always first shards data by time.

Segments contain data stored in compressed column orientations, along with the indexes for those columns. Druid queries only understand how to scan segments.
Following search infrastructure, Druid creates immutable snapshots of data, stored in data structures highly optimized for analytic queries.
Druid is a column store, which means each individual column is stored separately. Druid indexes data on a per-shard (segment) level.
Druid has two means of ingestion, real-time and batch. Real-time ingestion in Druid is best effort.
One common approach to operating Druid is to have a real-time pipeline for recent insights, and a batch pipeline for the accurate copy of the data.
Druid's native query language is JSON over HTTP. Apache Calcite - SQL parser, planner and query engine whose Druid adapter can query data residing in Druid.
Druid is designed to perform single table operations and does not currently support joins. Many production setups do joins at ETL because data must be denormalized before loading into Druid.
Druid is designed to have no single point of failure. Different node types are able to fail without impacting the services of the other node types.

A.Storage

• Segment:Druid倒排索引+时间分片

  Segments contain the various dimensions and metrics in a data set, stored in a column orientation, as well as the indexes for those columns.
  Segments are stored in a "deep storage" LOB store/file system.
  Druid stores its index in segment files, which are partitioned by time(按时间分片).
  Segment core data structure: three basic column types: the timestamp column, dimension columns, and metric columns.

Sharding Data to Create Segments

• Dimensions:Bitmap Index 维度:Bitmap索引

  Dimensions columns are different because they support filter and group-by operations, so each dimension requires the following three data structures:
  1.A dictionary that maps values (which are always treated as strings) to integer IDs,
  2.A list of the column’s values, encoded using the dictionary in 1
  3.For each distinct value in the column, a bitmap that indicates which rows contain that value.

The bitmaps in 3 -- also known as inverted indexes allow for quick filtering operations(specifically, bitmaps are convenient for quickly applying AND and OR operators).
The list of values in 2 is needed for group by and TopN queries.

• Multi-value columns

B.核心模块

• Broker模块:

route queries to if you want to run a distributed cluster. This node also merges the result sets from all of the individual nodes together.
Broker nodes employ a cache with a LRU cache invalidation strategy.
类似分布式搜索引擎中的meta元搜索引擎，他不负责任何Segment的查询，他只是一个代理，从Zookeeper中获取TimeLine，这个 TimeLine记录了intervals->List(Server)的mapping关系，接收到Client的请求以后，按照时间段在TimeLine查找Segment分布在那些 Server上。

• Coordinator模块:

responsible for loading new segments, dropping outdated segments, managing segment replication, and balancing segment load.
负责协调Segment的均衡分发加载，Coordinator从meta数据存储mysql/postgreSQL中获取那些还未被加载的Segment，根据当前所有Historical的负载能力均衡地分配到其LoadQueue。

• Historical模块:

从Deep Storage中下载Segment,采用mmap(内存映射)的方式加载Segment,并负责来自broker对这些Segment的查询.
Historical nodes do not communicate directly with each other or with the coordinator nodes but instead rely on Zookeeper for coordination.

• Indexing Service模块:

The indexing service is a highly-available, distributed service that runs indexing related tasks.Indexing service tasks create (and sometimes destroy) Druid segments.
The indexing service is composed of three main components: a peon component that can run a single task, a Middle Manager component that manages peons, and an overlord component that manages task distribution to middle managers.
Druid的索引结构布局由字典，正排(列存储)以及倒排索引组成，其中倒排的PostingList采用压缩LZ4的BitMap位图索引。支持Consice和Roaring两种BitMap方式

• Realtime process模块:

Realtime nodes will periodically build segments representing the data they’ve collected over some span of time and transfer these segments off to Historical nodes.

Realtime Node负责提供实时数据索引，生成realtime Index(Segment),并定期推送到Historical Node。在Realtime中采用LSM-Tree的模型

4.3.Druid Adapter & SQL Parser

Full SQL is currently not supported with Druid. (当前SQL支持无法与Druid原生查询语言一样灵活)
Calcite’s Druid adapter allows you to query the data using SQL, combining it with data in other Calcite schemas.

4.4.Multitenancy Consideration

Multitenant workloads can either use a separate datasource for each tenant, or can share one or more datasources between tenants using a "tenant_id" dimension. When deciding which path to go down, consider that each path has pros and cons.
Shared datasources or datasource-per-tenant
Druid offical compromise is to use more than one datasource, but a smaller number than tenants.
Partitioning shared datasources
Customizing data distribution
Supporting high query concurrency
Druid's fundamental unit of computation is a segment. Nodes scan segments in parallel and a given node can scan druid.processing.numThreads concurrently.
Druid internally stores requests to scan segments in a priority queue.

4.5.Query Caching

Druid supports query result caching through an LRU cache. Results are stored on a per segment basis, along with the parameters of a given query.

4.6.Sorting Order

These sorting orders are used by the TopNMetricSpec, SearchQuery, GroupByQuery's LimitSpec, and BoundFilter.

• Lexicographic
• Alphanumeric
• Numeric
• Strlen
• 不支持具体维度按度量排序功能