35-Spark Streaming反压机制、Spark的数据倾斜的解决和Kylin的简单介绍_spark streaming 反压

19.6 数据的反压机制⭐️

• 反压机制的理解
  • 数据流入的速度远高于数据处理的速度，对流处理系统构成巨大的负载压力，如果不能正确处理，可能导致集群资源耗尽，最终集群崩溃，
  • 因此有效的反压机制（backpressure）对保障流处理系统的稳定至关重要

19.6.1 Storm反压

1. 旧版本

1. 解决方案
   • 开启了acker机制的storm程序，可以通过设置conf.setMaxSpoutPending参数来实现反压效果，如果下游组件(bolt)处理速度跟不上导致spout发送的tuple没有及时确认的数超过了参数设定的值，spout会停止发送数据
2. 存在的问题
   • 但是conf.setMaxSpoutPending参数的设置很难达到最好的反压效果
     • 设小了会导致吞吐上不去
     • 设大了会导致worker OOM；有震荡，数据流会处于一个颠簸状态，效果不如逐级反压；
     • 另外对于关闭acker机制的程序无效；

2. 新版本

1. 解决方案

   • storm自动反压机制(Automatic Back Pressure)通过监控bolt中的接收队列的情况，
   • 当超过高水位值时专门的线程会将反压信息写到 Zookeeper ，Zookeeper上的watch会通知该拓扑的所有Worker都进入反压状态，最后Spout降低tuple发送的速度。

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vzgMEIVN-1658840308113)(C:/Users/18446/AppData/Roaming/Typora/typora-user-images/image-20220726183305228.png)]

19.6.2 Spark反压

1. 旧版本

1. 使用时机
   • Spark Streaming程序中当计算过程中出现batch processing time > batch interval的情况
     时，(其中batch processing time为实际计算一个批次花费时间，batch interval Streaming应用设置的批处理间隔)
2. 不处理的后果
   • 会导致数据在内存中堆积（OOM）或者 溢出到磁盘导致数据读取延迟加大
     • 意味着处理数据的速度小于接收数据的速度，如果这种情况持续过长的时间，会造成数据在内存中堆积，导致Receiver所在Executor内存溢出等问题
     • 如果设置StorageLevel包含disk, 则内存存放不下的数据会溢写至disk, 加大延迟
3. 解决方案
   • 可以通过设置参数spark.streaming.receiver.maxRate来限制Receiver的数据接收速率，
   • 虽然可以通过限制接收速率，来适配当前的处理能力，防止内存溢出，但也会引入其它问题
4. 存在的问题
   • 资源利用率下降
     • producer数据生产高于maxRate，当前集群处理能力也高于maxRate，这就会造成资
       源利用率下降等问题。

2. 新版本

1. 使用时机

   • Spark Streaming 从v1.5开始引入反压机制（back-pressure）,通过动态控制数据接收速率来适配集群数据处理能力
   • Spark Streaming Backpressure: 根据JobScheduler反馈作业的执行信息来动态调整Receiver数据接收率。

   

2. spark streaming的两种数据源

   • 若是基于Receiver的数据源，
     • 可以通过设置spark.streaming.receiver.maxRate来控制最大输入速率；
   • 若是基于Direct的数据源(如Kafka Direct Stream)，
     • 则可以通过设置spark.streaming.kafka.maxRatePerPartition 来控制最大输入速率。
   • 若设置spark.streaming.backpressure.enabled为true，Spark Streaming会自动根据处
     理能力来调整输入速率，从而在流量高峰时仍能保证最大的吞吐和性能。

3. 反压原理

   • 在原架构的基础上加上一个新的组件RateController,这个组件负责监听“OnBatchCompleted”事件，
   • 然后从中抽取processingDelay 及schedulingDelay信息.
   • Estimator依据这些信息估算出最大处理速度（rate），
   • 最后由基于Receiver的Input Stream将rate通过ReceiverTracker与ReceiverSupervisorImpl转发给BlockGenerator（继承自RateLimiter）

4. Spark Streaming的反压机制中的几个重要组件

   • RateController

     • RateController继承自接口StreamingListener,并实现了onBatchCompleted方
       法。每一个Batch处理完成后都会调用此方法

     override def onBatchCompleted(batchCompleted:
                                   StreamingListenerBatchCompleted) {
         val elements = batchCompleted.batchInfo.streamIdToInputInfo
         for {
             // 处理结束时间
             processingEnd <-
             batchCompleted.batchInfo.processingEndTime
             // 处理时间,即`processingEndTime` - `processingStartTime`
             workDelay <- batchCompleted.batchInfo.processingDelay
             // 在调度队列中的等待时间,即`processingStartTime` -
             `submissionTime`
             waitDelay <- batchCompleted.batchInfo.schedulingDelay
             // 当前批次处理的记录数
             elems <- elements.get(streamUID).map(_.numRecords)
         } computeAndPublish(processingEnd, elems, workDelay,
                             waitDelay)
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17

   • RateEstimator

     • RateEstimator是速率估算器，主要用来估算最大处理速率

     private def computeAndPublish(time: Long, elems: Long,
                                   workDelay: Long, waitDelay: Long): Unit = Future[Unit] {
         // 根据处理时间、调度时间、当前Batch记录数，预估新速率
         val newRate = rateEstimator.compute(time, elems, workDelay,
                                             waitDelay)
         newRate.foreach { s =>
             // 设置新速率
             rateLimit.set(s.toLong)
             // 发布新速率
             publish(getLatestRate())
         }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12

   • RateLimiter

     • RateLimiter是一个抽象类，它并不是Spark本身实现的，而是借助了第三方Google的GuavaRateLimiter来产生的
     • 它实质上是一个限流器，也可以叫做令牌，如果Executor中task每秒计算的速度大于该值则阻塞，如果小于该值则通过
     • 将流数据加入缓存中进行计算。这种机制也可以叫做令牌桶机制

     

     • 令牌桶机制

       数据流进入下一个阶段之前需要消耗令牌桶中的令牌，如果令牌桶中没有令牌，数据就暂时不能进入下一个阶段，只有令牌桶中有令牌时才可以进入下一个阶段

       • 大小固定的令牌桶可自行以恒定的速率源源不断地产生令牌。
       • 如果令牌不被消耗，或者被消耗的速度小于产生的速度，令牌就会不断地增多，直到把桶填满。
       • 后面再产生的令牌就会从桶中溢出。最后桶中可以保存的最大令牌数永远不会超过桶的大小。当进行某操作时需要令牌时会从令牌桶中取出相应的令牌数，如果获取到则继续操作，否则阻塞。用完之后不用放回。

5. 反压相关参数设置

   参数名称	默认值	说明
   spark.streaming.backpressure.enabled	false	是否启用反压机制
   spark.streaming.backpressure.initialRate	无	初始最大接收速率。只适用于Receiver Stream，不适用于Direct Stream。
   spark.streaming.backpressure.rateEstimator	pid	速率控制器,Spark 默认只支持此控制器，可自定义。
   spark.streaming.backpressure.pid.proportional	1.0	只能为非负值。当前速率与最后一批速率之间的差值对总控制信号贡献的权重。用默认值即可。
   spark.streaming.backpressure.pid.integral	0.2	只能为非负值。比例误差累积对总控制信号贡献的权重。用默认值即可
   spark.streaming.backpressure.pid.derived	0	只能为非负值。比例误差变化对总控制信号贡献的权重。用默认值即可
   spark.streaming.backpressure.pid.minRate	100	只能为正数，最小速率

   //启用反压
   conf.set("spark.streaming.backpressure.enabled","true")
   //最小摄入条数控制
   conf.set("spark.streaming.backpressure.pid.minRate","1")
   //最大摄入条数控制
   conf.set("spark.streaming.kafka.maxRatePerPartition","12")
   1
   2
   3
   4
   5
   6

6. 注意

   • 反压机制真正起作用时需要至少处理一个批：

     • 由于反压机制需要根据当前批的速率，预估新批的速率，所以反压机制真正起作用前，应至少保证处理一个批。

   • 如何保证反压机制真正起作用前应用不会崩溃：

     • 要保证反压机制真正起作用前应用不会崩溃,需要控制每个批次最大摄入速率。

     • 若为Direct Stream，如Kafka Direct Stream,则可以通过spark.streaming.kafka.maxRatePerPartition参数来控制。

       • 此参数代表了 每秒每个分区最大摄入的数据条数。

         • 假设BatchDuration为10秒，spark.streaming.kafka.maxRatePerPartition为12条，kafka topic 分区数为3个，则一个批(Batch)最大读取的数据条数为360条(31210=360)。

       • 该参数也代表了整个应用生命周期中的最大速率，即使是背压调整的最大值也不会超过
         该参数。

19.7 Spark Streaming事物

1. 对于事物的理解
   • Exactly-once 语义是实时计算的难点之一。要做到每一条记录只会被处理一次，即使服务器或网络发生故障时也能保证没有遗漏，这不仅需要实时计算框架本身的支持，还对上游的消息系统、下游的数据存储有所要求。
2. 一个Spark Streaming程序的三个组成部分
   • 输入、处理逻辑、输出。要达到exactly once的理想状态，需要三步协同进行，而不是只与处理逻辑有关。

19.7.1 逻辑处理

Spark 容错分为：Driver级别的容错和Executor级别的容错

1. Driver级别的容错

1. 出错位置
   • 在Driver级别的容错具体为DAG生成的模板
   • 即DStreamGraph，RecevierTracker中存储的元数据信息和JobScheduler中存储的Job进行的进度情况等信息
2. 对出错的处理：checkpoint
   • 只要通过checkpoint就可以了，每个Job生成之前进行checkpoint，在Job生成之后再进行checkpoint，如果出错的话就从checkpoint中恢复。

2. Executor级别的容错

1. 出错位置
   • 在Executor级别的容错具体为接收数据的安全性和任务执行的安全性
2. 对出错的处理–接收数据方面
   • 将数据保存到两台机器的内存中
     • 一种方式是Spark Streaming接收到数据默认为MEMORY_AND_DISK_2的方式，在两台机器的内存中，如果一台机器上的Executor挂了，立即切换到另一台机器上的Executor，这种方式一般情况下非常可靠且没有切换时间
   • 在处理数据前用WAL日志记录
     • 另外一种方式是WAL（Write Ahead Log），在数据到来时先通过WAL机制将数据进行日志记录，如果有问题则从日志记录中恢复，然后再把数据存到Executor中，再进行其他副本的复制
     • WAL这种方式对性能有影响，在生产环境中不常用，一般使用Kafka存储，Spark Streaming接收到数据丢失时可以从Kafka中回放。在任务执行的安全性方面，靠RDD的容错。
3. Spark Streaming的容错机制是基于RDD的容错机制
   • checkpoint
   • 基于血统（lineage）的高度容错机制
   • 出错了之后会从出错的位置从新计算，而不会导致重复计算

二十、Spark 优化

20.1 资源调优

20.2 并行度调优

20.3 代码调优

20.4 数据本地化

20.5 内存调优

20.6 Spark Shuffle调优

20.7 数据倾斜的解决

20.7.1 数据倾斜的问题产生

1. 理解数据倾斜
   • Spark 中的数据倾斜问题主要指 shuffle 过程中出现的数据倾斜问题，是由于不同的 key 对应的数据量不同导致的不同 task 所处理的数据量不同的问题。
   • 例如， reduce 点一共要处理 100 万条数据，第一个和第二个 task 分别被分配到了 1 万条数据， 计算 5 分钟内完成， 第三个 task 分配到了 98 万数据，此时第三个 task 可能需要 10 个小时完成，这使得整个 Spark 作业需要 10 个小时才能运行完成，这就是数据倾斜所带来的后果。
   • 注意，要区分开数据倾斜与数据量过量这两种情况，数据倾斜是指少数 task 被分配了绝大多数的数据， 因此少数 task 运行缓慢； 数据过量是指所有 task 被分配的数据量都很大， 相差不多， 所有 task 都运行缓慢。
2. 数据倾斜的表现
   • Spark 作业的大部分 task 都执行迅速，只有有限的几个 task 执行的非常慢，此时可能出现了数据倾斜， 作业可以运行，但是运行得非常慢；
   • Spark 作业的大部分 task 都执行迅速，但是有的 task 在运行过程中会突然报出 OOM，反复执行几次都在某一个 task 报出 OOM 错误，此时可能出现了数据倾斜，作业无法正常运行
3. 定位数据倾斜问题
   • 查阅代码中的 shuffle 算子，例如 reduceByKey、countByKey、groupByKey、 join 等算子，根据代码逻辑判断此处是否会出现数据倾斜；
   • 查看 Spark 作业的 log 文件， log 文件对于错误的记录会精确到代码的某一行，可以根据异常定位到的代码位置来明确错误发生在第几个 stage，对应的 shuffle算子是哪一个；

20.7.2 解决方案一：聚合原数据

1. 避免shuffle过程
   • 绝大多数情况下，Spark 作业的数据来源都是 Hive 表，这些 Hive 表基本都是经过 ETL 之后的昨天的数据。
   • 为了避免数据倾斜，我们可以考虑避免 shuffle 过程，如果避免了 shuffle 过程， 那么从根本上就消除了发生数据倾斜问题的可能。
2. 具体操作（数据源是Hive表）
   • 如果 Spark 作业的数据来源于 Hive 表， 那么可以先在 Hive 表中对数据进行聚合，
     • 例如按照 key 进行分组，将同一 key 对应的所有 value 用一种特殊的格式拼接到一个字符串里去，这样，一个 key 就只有一条数据了；
     • 之后，对一个 key 的所有 value进行处理时，只需要进行 map 操作即可，无需再进行任何的 shuffle 操作。通过上述方式就避免了执行 shuffle 操作，也就不可能会发生任何的数据倾斜问题。
   • 对于 Hive 表中数据的操作，不一定是拼接成一个字符串，也可以是直接对 key的每一条数据进行累计计算。

20.7.3 解决方案二：过滤导致倾斜的 key

如果在 Spark 作业中允许丢弃某些数据， 那么可以考虑将可能导致数据倾斜的 key 进行过滤，滤除可能导致数据倾斜的 key 对应的数据，这样，在 Spark 作业中就不会发生数据倾斜了。

20.7.4 解决方案三：提高 shuffle 操作中的 reduce 并行度

1. 实现原理
   • 当方案一和方案二对于数据倾斜的处理没有很好的效果时，可以考虑提高shuffle过程中的reduce端并行度，reduce端并行度的提高就增加了reduce端task的数量，那么每个task分配到的数量就会相应的减少，由此缓解数据倾斜问题
2. reduce端并行度的设置
   • 在大部分的 shuffle 算子中， 都可以传入一个并行度的设置参数， 比如reduceByKey(500)，这个参数会决定 shuffle 过程中 reduce 端的并行度，在进行 shuffle操作的时候，就会对应着创建指定数量的 reduce task。
   • 对于 Spark SQL 中的 shuffle类语句，比如 group by、join 等，需要设置一个参数，即 spark.sql.shuffle.partitions，该参数代表了 shuffle read task 的并行度，该值默认是 200， 对于很多场景来说都有点过小。
3. 增加 reduce read task 的数量
   • 增加 shuffle read task 的数量，可以让原本分配给一个 task 的多个 key 分配给多个 task，从而让每个 task 处理比原来更少的数据。
   • 举例来说，如果原本有 5 个 key，每个 key 对应 10 条数据，这 5 个 key 都是分配给一个 task 的，那么这个 task 就要处理 50 条数据。而增加了 shuffle read task 以后，每个 task 就分配到一个 key，即每个 task 就处理 10 条数据， 那么自然每个 task 的执行时间都会变短了。
4. reduce 端并行度设置存在的缺陷
   • 只是缓解了shuffle的压力，提高了运行速度，并没有从根本上解决shuffle问题
     • 提高 reduce 端并行度并没有从根本上改变数据倾斜的本质和问题（ 方案一和方案二从根本上避免了数据倾斜的发生）， 只是尽可能地去缓解和减轻 shuffle reduce task 的数据压力，以及数据倾斜的问题， 适用于有较多 key 对应的数据量都比较大的情况。
   • 发现数据倾斜尝试的第一种手段，无法根本上解决数据倾斜（无法处理一个key对应大量的数据）
     • 该方案通常无法彻底解决数据倾斜，因为如果出现一些极端情况，比如某个 key对应的数据量有 100 万，那么无论你的 task 数量增加到多少，这个对应着 100 万数据的 key 肯定还是会分配到一个 task 中去处理，因此注定还是会发生数据倾斜的。
     • 所以这种方案只能说是在发现数据倾斜时尝试使用的第一种手段， 尝试去用最简单的方法缓解数据倾斜而已，或者是和其他方案结合起来使用。
   • 提高reduce端并行度的效果不明显后，需及时更换其他方案
     • 在理想情况下，reduce 端并行度提升后，会在一定程度上减轻数据倾斜的问题，甚至基本消除数据倾斜；但是，在一些情况下，只会让原来由于数据倾斜而运行缓慢的 task 运行速度稍有提升， 或者避免了某些 task 的 OOM 问题， 但是， 仍然运行缓慢，此时，要及时放弃方案三，开始尝试后面的方案。

20.7.5 解决方案四：使用随机 key 实现双重聚合

1. 理解

   • 当使用了类似于 groupByKey、reduceByKey 这样的算子时，可以考虑使用随机

     key 实现双重聚合，如图所示：

   

2. 实现步骤

   • 首先， 通过 map 算子给每个数据的 key 添加随机数前缀， 对 key 进行打散， 将原先一样的 key 变成不一样的 key，
   • 然后进行第一次聚合， 这样就可以让原本被一个 task 处理的数据分散到多个 task 上去做局部聚合；
   • 随后，去除掉每个 key 的前缀， 再次进行聚合，得到最终结果

3. 具体实现步骤

   这个方案的核心实现思路就是进行两阶段聚合。

   • 第一次是局部聚合，先给每个key都打上一个随机数，比如10以内的随机数，此时原先一样的key就变成不一样的了，比如(hello, 1) (hello, 1) (hello, 1) (hello,1)，就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。
   • 接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合，那么局部聚合结果，就会变成了(1_hello, 2) (2_hello, 2)。
   • 然后将各个key的前缀给去掉，就会变成(hello,2)(hello,2)，再次进行全局聚合操作，就可以得到最终结果了，比如(hello, 4)。

4. 适用场景

   • 此方法对于由 groupByKey、reduceByKey 这类算子造成的数据倾斜由比较好的效果，仅仅适用于聚合类的 shuffle 操作，适用范围相对较窄。如果是 join 类的 shuffle操作，还得用其他的解决方案
   • 此方法也是前几种方案没有比较好的效果时要尝试的解决方案

20.7.6 解决方案五：将reduce join 转换为 map join

1. 理解 reduce join

   • 正常情况下， join 操作都会执行 shuffle 过程，并且执行的是 reduce join，也就是先将所有相同的 key 和对应的 value 汇聚到一个 reduce task 中，然后再进行 join。普通 join 的过程如下图所示：

   

   • 普通的 join 是会走 shuffle 过程的，而一旦 shuffle，就相当于会将相同 key 的数据拉取到一个 shuffle read task 中再进行 join， 此时就是 reduce join。

2. 对于一个较小的RDD的处理思路

   • 如果一个 RDD 是比较小的，则可以采用广播小 RDD 全量数据 +map 算子来实现与 join 同样的效果，也就是 map join，此时就不会发生 shuffle 操作，也就不会发生数据倾斜。
   • 注意
     • RDD 是并不能进行广播的，只能将 RDD 内部的数据通过 collect 拉取到 Driver 内存然后再进行广播

3. 实现思路

   • 不使用 join 算子进行连接操作， 而使用 Broadcast 变量与 map 类算子实现 join操作，进而完全规避掉 shuffle 类的操作，彻底避免数据倾斜的发生和出现。

4. 实现步骤

   • 将较小 RDD 中的数据直接通过 collect 算子拉取到 Driver 端的内存中来，然后对其创建一个 Broadcast 变量；
   • 接着对另外一个 RDD 执行 map 类算子， 在算子函数内， 从 Broadcast 变量中获取较小 RDD 的全量数据，与当前 RDD 的每一条数据按照连接 key 进行比对，如果连接 key 相同的话，那么就将两个 RDD 的数据用你需要的方式连接起来。

5. 适用场景

   • 当 join 操作有数据倾斜问题并且其中一个 RDD 的数据量较小时， 可以优先考虑这种方式，效果非常好。map join 的过程如图所示：

   

6. 不适合场景分析

   • RDD数据量较大
     • 由于 Spark 的广播变量是在每个 Executor 中保存一个副本，如果两个 RDD 数据量都比较大， 那么如果将一个数据量比较大的RDD 做成广播变量，那么很有可能会造成内存溢出。

20.7.7 解决方案六：sample 采样对倾斜 key 单独进行 join

20.7.8 解决方案七：使用随机数以及扩容进行 join

1. 对于RDD中大量的key导致的数据倾斜用扩容解决

   • 如果在进行 join 操作时，RDD 中有大量的 key 导致数据倾斜，那么进行分拆 key也没什么意义，此时就只能使用最后一种方案来解决问题了， 对于 join 操作，我们可以考虑对其中一个 RDD 数据进行扩容， 另一个 RDD 进行稀释后再 join。

2. 解决思路

   • 大量倾斜的key 用对key进行随机数分散到不同task + 对不能将拆分的key进行单独的处理（扩容整个RDD）
     • 我们会将原先一样的 key 通过附加随机前缀变成不一样的 key，然后就可以将这些处理后的“不同 key”分散到多个 task 中去处理，而不是让一个 task 处理大量的相同 key。这一种方案是针对有大量倾斜 key 的情况，没法将部分 key 拆分出来进行单独处理， 需要对整个 RDD 进行数据扩容， 对内存资源要求很高。

3. 实现步骤

   • 选择一个 RDD，使用 flatMap 进行扩容，对每条数据的 key 添加数值前缀（ 1~N的数值）， 将一条数据映射为多条数据；（扩容）
   • 选择另外一个 RDD，进行 map 映射操作，每条数据的 key 都打上一个随机数作为前缀（1~N 的随机数）；（稀释）
   • 将两个处理后的 RDD，进行 join 操作。

   

4. 局限性

   • 如果两个 RDD 都很大，那么将 RDD 进行 N 倍的扩容显然行不通；使用扩容的方式只能缓解数据倾斜，不能彻底解决数据倾斜问题。

5. 使用方案七对方案六进一步优化分析

   当 RDD 中有几个 key 导致数据倾斜时，方案六不再适用，而方案七又非常消耗资源，此时可以引入方案七的思想完善方案六

   1. 对包含少数几个数据量过大的 key 的那个 RDD，通过 sample 算子采样出一份样本来，然后统计一下每个 key 的数量，计算出来数据量最大的是哪几个 key。

   2. 然后将这几个 key 对应的数据从原来的 RDD 中拆分出来， 形成一个单独的 RDD，并给每个 key 都打上 n 以内的随机数作为前缀，而不会导致倾斜的大部分 key形成另外一个 RDD。

   3. 接着将需要 join 的另一个 RDD， 也过滤出来那几个倾斜 key 对应的数据并形成一个单独的 RDD，将每条数据膨胀成 n 条数据，这 n 条数据都按顺序附加一个 0~n 的前缀， 不会导致倾斜的大部分 key 也形成另外一个 RDD。

   4. 再将附加了随机前缀的独立 RDD 与另一个膨胀 n 倍的独立 RDD 进行 join，此时就可以将原先相同的 key 打散成 n 份， 分散到多个 task 中去进行 join 了。

   5. 而另外两个普通的 RDD 就照常 join 即可。

      最后将两次 join 的结果使用 union 算子合并起来即可，就是最终的 join 结果

20.8 Spark故障解决

20.2 并行度调优

二十一、Kylin

21.1 Kylin简介

21.1.1 Kylin由来

1. Kylin介绍

   • Apache Kylin （ Extreme OLAP Engine for Big Data ）是一个开源的分布式分析引擎，为 Hadoop等大型分布式数据平台之上的超大规模数据集通过标准 SQL 查询及多维分析（ OLAP ）功能，提供亚秒级的交互式分析能力
   • 离线分析引擎，但是有实时的效果，因为数据已经提前算好了（将能算出来的数据都算出来）

2. Kylin是一个框架的集合

   • kylin：Hive+HBASE+Hadoop+Spark+kafka
   • Kylin不会计算历史数据的总和，会计算一个周期的数据，
     • 如前一天的数据，前一个月的数据，不会计算从数据产生以来到现在为止的数据

21.1.2 Kylin优势

Apache Kylin 的主要特点包括支持 SQL 接口、支持超大数据集、亚秒级响应、可伸缩性、高吞吐率、
BI 工具集成等

1. 标准SQL接口

2. 支持超大规模集

因为使用了 Cube 预计算技术，在理论上， Kylin 可以支撑的数据集大小没有上限，仅受限于存储系
统和分布式计算系统的承载能力，并且查询速度不会随数据集的增大而减慢。 Kylin 在数据集规模上的
局限性主要在于维度的个数和基数。它们一般由数据模型来决定，不会随着数据规模的增长而线性增
长，这也意味着 Kylin 对未来数据的增长有着更强的适应能力、

3. 亚秒级相应

预计算

Apache Kylin 拥有优异的查询响应速度，这点得益于预计算，很多复杂的计算，比如连接、聚合，在
离线的预计算过程中就已经完成，这大大降低了查询时刻所需要的计算量，提高了响应速度。

4. BI 及可视化工具集成

Apache Kylin 提供了丰富的 API ，以与现有的 BI 工具集成，具体包括如下内容。
ODBC 接口：与 Tableau 、 Excel 、 Power BI 等工具集成。
JDBC 接口：与 Saiku 、 BIRT 等 Java 工具集成。
Rest API ：与 JavaScript 、 Web 网页集成。

21.1.3 Kylin工作原理⭐️

1. Kylin的工作原理：
   • 典型的空间换时间、利用cube计算的结构加速我们的查询
2. 具体过程
   • 指定数据模型，定义维度和度量。
   • 预计算 Cube ，计算所有 Cuboid 并保存为物化视图。
   • 执行查询时，读取 Cuboid ，运算，产生查询结果
3. 预计算
   • Kylin 的查询过程不会扫描原始记录，而是通过预计算预先完成表的关联、聚合等复杂运算，并利用预计算的结果来执行查询，因此相比非预计算的查询技术，其速度一般要快一到两个数量级，并且这点在超大的数据集上优势更明显。当数据集达到千亿乃至万亿级别时， Kylin 的速度甚至可以超越其他非预计算技术1000倍以上
4. 总结
   • Kylin 的核心思想是 Cube 预计算，理论基础是空间换时间，把高复杂度的聚合运算、多表连接等操作转换成对预计算结果的查询

21.1.4 Kylin技术架构⭐️

21.1.5 Kylin总结

1. 传统技术
   • 大规模的并行计算和列式存储的查询速度都在O(n)级别，与数据规模增长呈线性关系
   • 如果数据规模增长10倍，那么 O(N) 的查询速度就会下降到十分之一，无法满足日益增长的数据需求。
2. Kylin
   • 依靠 Apache Kylin ，我们不用再担心查询速度会随着数据量的增长而减慢