为云计算带来交互式

　　基于Hadoop的SQL一直在被持续地改进，但是一个查询要等几分钟到几小时还是非常得正常。在这篇博文里，我们将会介绍开源的分布式分析引擎Apache Kylin。重点介绍它是如何以数量级加速大数据查询，以及在2.0版里面为交互式BI所提供的新特性，包括对雪花模型的支持和流式建立数据立方。本篇文章给大家带来的就是关于数据的一些详细解析，本篇文章会教给大家数据的知识点进行分析，希望本篇文章能帮助到你，对你有所收获，希望大家仔细阅读文章。

　　Apache Kylin是什么?

　　Kylin是一个在Hadoop上的OLAP引擎，Kylin位于Hadoop之上，向上层的应用提供了基于标准SQL接口的关系型数据。

　　Kylin可以处理大数据集，从查询延迟上讲是很快的，这也是它和其他基于Hadoop的SQL的区别。比如，我们所知道的使用Kylin的最大的生产系统实例是在今日头条，一个中国的新闻推送应用。头条有一个包含3万亿条记录的表，对它的平均查询响应时间低于1秒。下一节我们会讨论Kylin是怎么实现这么快的查询。

　　Kylin引擎的另一个特点是它可以支持复杂的数据模型。 例如，太平洋保险(CPIC，中国的一个保险集团公司)有一个多达60维的模型。 Kylin提供标准的JDBC / ODBC / RestAPI接口，可实现与任何SQL应用程序的连接。

　　Kyligence还开发了一个在线演示，展示了使用10亿条航班记录的BI体验。你可以查看这里来了解。比如，在旧金山国际机场过去20年里延误最久的航班是哪个。(使用用户名“analyst”和密码“analyst”登录，选择“airline_cube”，拖放维度和事实数据来查询这个数据集)

　　一个零售业场景：展示Kylin的速度

　　Kylin比传统的基于Hadoop的SQL要快，是因为它采用了预计算来在SQL执行前先行一步。例如，设想一个零售业务场景，你需要处理非常多的订单，每个订单包含多个商品。如果想知道订单取消和退货造成的影响，一个分析人员可能需要写一个查询来在某个时间段内按照“returnflas(退货标记)”和“orderstatus(订单状态)”来汇总收入进行汇报，如图2 所示。图里面显示了这个查询被编译成的关系表达式，也叫执行计划。

　　从这个执行计划可以很容易地看出，这个执行的时间复杂度至少是O(N)，这里N是表里的总行数，因为每行都要至少被访问一次。同时我们假定要关联的表都已经很好地被分区和索引过了，因此花费比较大的关联操作也可以在线性的时间复杂度上完成，但在实际场景里这种情况是不大可能的。

　　这个O(N)的时间复杂度并不好，因为这意味着如果数据量增长十倍，则查询时间也会慢10倍。现在一个查询需要花1秒钟，那么以后随着数据的增长，这个时间会变成10秒甚至是100秒。我们想要的是无论数据量大小，这个查询时间都是固定不变的。

　　Kylin的解决方法是预计算。整体思路是，如果我们提前知道查询的模式，我们就能预先计算出整个执行的一部分。在上面这个例子里，就是预先计算Aggregate、Join和表扫描操作。生成的结果是一个立方体理论里的数据立方(或者可以把它叫做“实例化的总结表”，如果这样听起来觉得更好的话)。

　　最初的执行计划就被转换成了基于数据立方之上。这个数据立方体包含了按照“returnflag(退货标记)”、“orderstatus(订单状态)”和“date(日期)”进行汇总的记录。因为退货标记和订单状态是一个固定的数量，而日期范围被限定在3年(大概1000天)。这就意味着这个数据立方体里的行数最多就是“标记数×状态数×天数”，对O定义的时间复杂度来说就是一个常量。这个新的执行计划将会保证不管源数据有多大都有一个固定的执行时间。这就我们想要的效果!

　　Kylin的架构一览

　　如我们所见，Kylin是一个依赖于预计算的系统。其核心是基于经典的立方体理论，并发展成一个大数据上的SQL解决方案(见图4)。它使用模型和立方体概念来定义预计算的空间。构建引擎从数据源载入数据，并在使用MapReduce或Spark的分布式系统上进行预计算。被计算出来的立方体被存储在HBase里。当一个查询来到时，Kylin把它编译成一个关系表达式，找到匹配的模型，并基于预计算好的数据立方体而不是原始数据执行这个表达式

　　这里的关键就是建模。如果你对数据以及分析的需求有非常好的理解，你是可以用正确的模型和立方体定义来找到正确的预计算。接着，绝大多数(如果不是全部)的查询都可以被转化成对此立方体的查询。如图5所示，执行时间复杂度可以被降低到O(1)，从而获得非常快的查询速度，无论原始数据有多大。

　　Kylin 2.0的特性

　　对雪花模型的支持和TPC-H基准测试

　　在Storm出现之前，进行实时处理是非常痛苦的事情，我们主要的时间都花在关注往哪里发消息，从哪里接收消息，消息如何序列化，真正的业务逻辑只占了源代码的一小部分。一个应用程序的逻辑运行在很多worker上，但这些worker需要各自单独部署，还需要部署消息队列。最大问题是系统很脆弱，而且不是容错的：需要自己保证消息队列和worker进程工作正常。

　　Storm完整地解决了这些问题。它是为分布式场景而生的，抽象了消息传递，会自动地在集群机器上并发地处理流式计算，让你专注于实时处理的业务逻辑。

　　Storm有如下特点：

　　1) 编程简单：开发人员只需要关注应用逻辑，而且跟Hadoop类似，Storm提供的编程原语也很简单

　　2) 高性能，低延迟：可以应用于广告搜索引擎这种要求对广告主的操作进行实时响应的场景。

　　3) 分布式：可以轻松应对数据量大，单机搞不定的场景

　　4) 可扩展：随着业务发展，数据量和计算量越来越大，系统可水平扩展

　　5) 容错：单个节点挂了不影响应用

　　6) 消息不丢失：保证消息处理

　　不过Storm不是一个完整的解决方案。使用Storm时你需要关注以下几点：

　　1) 如果使用的是自己的消息队列，需要加入消息队列做数据的来源和产出的代码

　　2) 需要考虑如何做故障处理：如何记录消息处理的进度，应对Storm重启，挂掉的场景

　　3) 需要考虑如何做消息的回退：如果某些消息处理一直失败怎么办?

　　2. Storm与Hadoop区别

　　1) 定义及架构

　　Hadoop是Apache的一个项目，是一个能够对大量数据进行分布式处理的软件框架。

　　Storm是Apache基金会的孵化项目，是应用于流式数据实时处理领域的分布式计算系统。

　　HadoopStorm

　　系统角色JobTrackerNimbus

　　TaskTrackerSupervisor

　　ChildWorker

　　应用名称JobTopology

　　组件接口Mapper/ReducerSpout/Bolt

　　2) 应用方面

　　Hadoop是分布式批处理计算，强调批处理，常用于数据挖掘和分析。

　　Storm是分布式实时计算，强调实时性，常用于实时性要求较高的地方。

　　3) 计算处理方式

　　Hadoop是磁盘级计算，进行计算时，数据在磁盘上，需要读写磁盘;Hadoop应用MapReduce的思想，将数据切片计算来处理大量的离线数据。Hadoop处理的数据必须是已经存放在HDFS上或者类似HBase的数据库中，所以Hadoop实现的时候是通过移动计算到这些存放数据的机器上来提高效率的。

　　Storm是内存级计算，数据直接通过网络导入内存。Storm是一个流计算框架，处理的数据是实时消息队列中的，需要写好一个Topology逻辑，然后将接收进来的数据进行处理，所以Storm是通过移动数据平均分配到机器资源来获得高效率的。

　　4) 数据处理方面

　　数据来源：Hadoop是HDFS上某个文件夹下的数据，数据量可能以TB来计;而Storm则是实时新增的某一笔数据。

　　处理过程：Hadoop是Map阶段到Reduce阶段的;Storm是由用户定义处理流程，流程中可以包含多个步骤，每个步骤可以是数据源(SPOUT)，也可以是处理逻辑(BOLT)。

　　是否结束：Hadoop最后必须要结束;而Storm没有结束状态，到最后一步时，就停在那，直到有新数据进入时再重新开始。

　　处理速度：Hadoop以处理HDFS上大量数据为目的，速度慢;Storm只要处理新增的某一笔数据即可，故此它的速度很快。

　　适用场景：Hadoop主要是处理一批数据，对时效性要求不高，需要处理就提交一个JOB;而Storm主要是处理某一新增数据的，故此时效性要求高。

　　总结，Hadoop和Storm并没有真的优劣之分，它们只是在各自的领域上有着独特的性能而已，若是真的把它们进行单纯的比较，反而是有失公平了。事实上，只有在最合适的方面使用最合适的大数据平台，才能够真正体现出它们的价值，也才能够真正为我们的工作提供最为便捷的助力!

　　3. Storm基本概念

　　1) Topology

　　一个Storm拓扑打包了一个实时处理程序的逻辑。一个Storm拓扑跟一个MapReduce的任务(job)是类似的。主要区别是MapReduce任务最终会结束，而拓扑会一直运行(当然直到你杀死它)。一个拓扑是一个通过流分组(Stream Grouping)把Spout和Bolt连接到一起的拓扑结构。图的每条边代表一个Bolt订阅了其他Spout或者Bolt的输出流。一个拓扑就是一个复杂的多阶段的流计算

　　2) Tuple

　　元组是Storm提供的一个轻量级的数据格式，可以用来包装你需要实际处理的数据。元组是一次消息传递的基本单元。一个元组是一个命名的值列表，其中的每个值都可以是任意类型的。元组是动态地进行类型转化的—字段的类型不需要事先声明。在Storm中编程时，就是在操作和转换由元组组成的流。通常，元组包含整数，字节，字符串，浮点数，布尔值和字节数组等类型。要想在元组中使用自定义类型，就需要实现自己的序列化方式。

　　3) Stream

　　流是Storm中的核心抽象。一个流由无限的元组序列组成，这些元组会被分布式并行地创建和处理。通过流中元组包含的字段名称来定义这个流。

　　每个流声明时都被赋予了一个ID。只有一个流的Spout和Bolt非常常见，所以OutputFieldsDeclarer提供了不需要指定ID来声明一个流的函数(Spout和Bolt都需要声明输出的流)。这种情况下，流的ID是默认的“default”。

　　4) Spout

　　Spout(喷嘴，这个名字很形象)是Storm中流的来源。通常Spout从外部数据源，如消息队列中读取元组数据并吐到拓扑里。Spout可以是可靠的(reliable)或者不可靠(unreliable)的。可靠的Spout能够在一个元组被Storm处理失败时重新进行处理，而非可靠的Spout只是吐数据到拓扑里，不关心处理成功还是失败了。

　　Sput可以一次给多个流吐数据。此时需要通过OutputFieldsDeclarer的declareStream函数来声明多个流并在调用SpoutOutputCollector提供的emit方法时指定元组吐给哪个流。

　　Spout中最主要的函数是nextTuple，Storm框架会不断调用它去做元组的轮询。如果没有新的元组过来，就直接返回，否则把新元组吐到拓扑里。nextTuple必须是非阻塞的，因为Storm在同一个线程里执行Spout的函数。

　　Spout中另外两个主要的函数是Ack和fail。当Storm检测到一个从Spout吐出的元组在拓扑中成功处理完时调用Ack,没有成功处理完时调用Fail。只有可靠型的Spout会调用Ack和Fail函数。

　　5) Bolt

　　在拓扑中所有的计算逻辑都是在Bolt中实现的。一个Bolt可以处理任意数量的输入流，产生任意数量新的输出流。Bolt可以做函数处理，过滤，流的合并，聚合，存储到数据库等操作。Bolt就是流水线上的一个处理单元，把数据的计算处理过程合理的拆分到多个Bolt、合理设置Bolt的task数量，能够提高Bolt的处理能力，提升流水线的并发度。

　　Bolt可以给多个流吐出元组数据。此时需要使用OutputFieldsDeclarer的declareStream方法来声明多个流并在使用[OutputColletor]的emit方法时指定给哪个流吐数据。

　　当你声明了一个Bolt的输入流，也就订阅了另外一个组件的某个特定的输出流。如果希望订阅另一个组件的所有流，需要单独挨个订阅。InputDeclarer有语法糖来订阅ID为默认值的流。例如declarer.shuffleGrouping("redBolt")订阅了redBolt组件上的默认流，跟declarer.shuffleGrouping("redBolt", DEFAULT_STREAM_ID)是相同的。

　　在Bolt中最主要的函数是execute函数，它使用一个新的元组当作输入。Bolt使用OutputCollector对象来吐出新的元组。Bolts必须为处理的每个元组调用OutputCollector的ack方法以便于Storm知道元组什么时候被各个Bolt处理完了(最终就可以确认Spout吐出的某个元组处理完了)。通常处理一个输入的元组时，会基于这个元组吐出零个或者多个元组，然后确认(ack)输入的元组处理完了，Storm提供了IBasicBolt接口来自动完成确认。

　　必须注意OutputCollector不是线程安全的，所以所有的吐数据(emit)、确认(ack)、通知失败(fail)必须发生在同一个线程里。更多信息可以参照问题定位

　　6) Task

　　每个Spout和Bolt会以多个任务(Task)的形式在集群上运行。每个任务对应一个执行线程，流分组定义了如何从一组任务(同一个Bolt)发送元组到另外一组任务(另外一个Bolt)上。可以在调用TopologyBuilder的setSpout和setBolt函数时设置每个Spout和Bolt的并发数。

　　Kylin 2.0引入了对元数据建模的增强，并且可以支持开箱即用的雪花模型。为了演示建模和SQL能力上的改进，我们进行了用Kylin 2.0运行TPC-H查询的基准测试。

　　需要注意的是，这里的目标并不是想与其他人的TPC-H结果进行比较。一方面，根据TPC-H规范，不允许在表之间进行预先计算，因此在这个意义上，Kylin不能算是有效的TPC-H系统。另一方面，我们还没有对这些查询进行性能调优。改善的空间还是很大的。

　　根据相同的零售场景，让我们快速查看一些有趣的TPC-H查询。图6是TPC-H查询07。SQL里面的字太小，可能看不清楚，但它给出了查询的复杂性的粗略感觉。该图是执行计划，强调了预计算(白色)与在线计算(蓝色)的部分。很容易看出，大部分关系运算符是预先计算的。剩下的Sort / Proj / Filter在很少的记录上工作，因此查询可以超快。在相同的硬件和相同的数据集上，Kylin用了0.17秒完成，而Hive + Tez对此查询运行了35.23秒。这显示了预计算所带来的差异。

　　这个查询有四个子查询，比前一个更复杂。 其执行计划包括两个分支，每个分支从预先计算的立方体载入数据。 分支结果再连接起来，这是一个复杂的在线计算。随着在线计算部分的增加，预计算的部分减少，Kylin的运行时间更长：3.42秒。 然而，完全在线计算的Hive + Tez仍然要慢一点，运行时间为15.87秒。

　　为近实时分析准备的流式立方体

　　预先计算给ETL流程增加了额外的时间，这在实时场景中会成为一个问题。为了解决这个问题，Kylin现在支持增量加载新添加的数据，而不会影响历史数据。 已有RestAPI可用于自动触发增量构建。每日构建一次是最常见的，现在更频繁的数据加载也是可行的。

　　从1.6版开始，Kylin可以直接从Kafka获取数据，并进行近乎实时的数据分析。使用基于内存的立方体算法，微型增量构建可以在几分钟内完成。生成的结果是许多小的立方体分片，可以给查询提供实时的结果。

　　为了展示这个特性是如何运作的，我们构建了一个演示网站来实时分析Twitter消息。它运行在一个八个节点的AWS集群上，有三个Kafka broker。输入是一个Twitter样本源，每秒有超过10K条消息。立方体的平均复杂度是：九个维度和三个测量数据。增量构建是每两分钟触发一次，并在三分钟内完成。总体而言，系统在实时性上有五分钟的延迟。

　　该演示按照语言和设备维度显示了Twitter消息的趋势。在图8中可以看到，美国白天的英文消息量上升，而亚洲语言的消息量在夜间下降。演示里还有一个标签云，用以显示最新的热门话题。在标签云下面是最热门标签的趋势。所有图表都是实时到最近五分钟。

　　总结

　　Apache Kylin是Hadoop上一个流行的OLAP引擎。通过使用预计算技术，它可以使SQL对大数据的查询速度有数量级的加快，并使交互式BI和其他在线应用程序能够直接在大数据上运行。

　　Kylin 2.0是最新版本，可以在这里下载。新版本的特性包括：Hadoop上的小于秒级的SQL延迟;雪花模型的支持和成熟的SQL功能;流式立方体用于近实时分析;内置时间/窗口/百分位数功能;和可以将构建时间缩短一半的Spark构建立方体。想要学习更多知识，那就来课课家教育，我们这里有通俗易懂的噢~不怕你们学不会!你的支持就是课课家教育最大的动力，欢迎进入课课家教育!