基于KUDU的实时数仓平台架构实践

作者：Kim来源：瞎掰数据智能
目录

• 业务背景
• 架构演进
• 技术分解
• 调优实践
  
  
  

“数据智能” (Data Intelligence) 有一个必须且基础的环节，就是数据仓库的建设，同时，数据仓库也是公司数据发展到一定规模后必然会提供的一种基础服务。从智能商业的角度来讲，数据结果的及时性就显得尤为重要，快速的获取数据反馈能够帮助公司更快的做出决策，更好的进行产品迭代，实时数仓在这一过程中起到了不可替代的作用。今天我将详细介绍QD大数据平台重要的一个环节-实时数仓平台。

业务背景

当前集团业务实时场景需求增多，现有技术栈已不能满足，如大数据量下实时数据写入/更新、多维数据分析、运营根据用户实时状态筛选用户进行营销等需求，使得可以在秒级针对亿级数据进行即时的多维分析透视和业务探索，能让用户在瞬息之间进行灵活的数据探索，快速发现数据价值。
业务数据特点及数据场景需求如下：集群规模：各类存储计算集群共300+个节点（20TB+内存、7000+C）数据规模：累计热数据4PB+，每日新增30TB+，每日处理2.5万+次任务数据需求：

• 刚刚到达的数据就马上要被终端用户使用访问到
• 支持在海量数据中做明细查询和快速响应结果
• 基于历史数据及最新数据进行实时决策
• 支持流SQL及批SQL开发数据，在成本与效率目标下达到数据口径的统一
  
  

场景需求：实时报表、实时用户画像、实时营销、实时流量、多维分析、AD-HOC

现状&痛点

• 实时存储：如何实时插入、更新、删除数据？
• 实时计算：数据什么时候能跑完？什么时候能看到？
  
  
  
  

架构目标

• 实时计算/存储
• 实时与离线数据一致性
• 业务与技术解耦
• 体系化-可复用、配套建设完善
  
  

业界方案

一个成熟的实时数据仓库平台架构包含以下环节：

• 实时数据采集（flume、logstash、filebeat）
• 实时数据交换（storm、Sparkstreaming、flink）
• 数据存储服务（hbase、druid、es、kudu、deltalake、kafka）
• 任务调度系统（Oozie、azkaban、Airflow）
• 实时任务计算引擎（impala、spark、flink）
  
  

其中，实时数据存储服务作为最底层的基础服务，技术选型至关重要。下面我们选择hdfs、hbase和kudu做下一些特性分析：

Kudu VS Hbase

Hbase:

列扫描，会读到其他列数据

基于LSM-Tree，读取数据需要合并多版本文件以及memstore

Kudu:

索引基于B树，加速查找

一个key只会存在一个rowset中，每个rowset记录key范围

其他：如Druid 预聚合特性、不支持精确去重、append only；clickhouse 不支持update/delete、不支持事务；类MySQL数据库 扩展性低、成本高等。

最终我们决定采用Kudu作为我们的实时数仓底层存储方案，它有以下几种特性：

• 支持随机读写：接近hbase的吞吐性能、更低的延迟（1ms ssd）
• 高性能的顺序读：接近HDFS的读性能
• 高可用：Raft一致性协议
• 类关系型数据库数据模型：支持行级事务、计算引擎支持Impala/Spark等sql api
  

架构演进

实时体系建设的三个阶段：

1. storm实时计算（2017.6）

处理流程：storm ->mysql -> 可视化、实时报表

主要是对流量数据做实时 ETL，case by case计算实时指标，但未建立起实时数仓体系，实时场景比较单一，对实时数据流的处理主要是为了提升数据平台的服务能力。实时数据的处理向上依赖数据的收集，向下关系到数据的查询和可视化

2. Flink实时营销 （2018.12）

处理流程：dts->kafka->flink -> config server -> redis/mysql -> 实时营销

随着业务的增长，各业务消费实时数据的需求开始增多，case by case烟囱式开发模式，带来了开发及维护成本，为适应灵活快速的业务变化，在原实时计算基础上，增加业务数据订阅功能和配置化服务，通过用户群配置和营销活动配置，实现精准实时营销。

3. Kudu实时数仓 （2019.8）

处理流程：dts->kafka->flink->kudu -> 实时数据仓库

借鉴已有的成熟的离线数仓体系，结合实时计算，梳理出了新的架构设计并开始着手建立实时数仓体系。

实时数仓建设思路

数仓建设的本质目的是支撑分析决策，那分析决策依赖什么样的数据，数仓建设是如何保障这些数据高效正确产出的。面向业务数据指标建设数仓，同时兼顾其它可能的扩展情况，是数仓建设的整体思路。数仓对上层数据应用的支持主要体现在三个方面：业务监控数据（大盘数据洞察）、数据挖掘(用户画像、推荐等）、数据分析(业务诊断、提升优化等)。其中按支持的优先级来分，首先就是业务监控数据，然后是数据分析、再然后数据挖掘。这也是数据应用由浅到深的递进。

从数仓的组成上来分以下几个模块进行数仓的建设：

• 数仓规划策略：分层、分线、分主题（数据集市）
• 数仓模型思想：ER模型、维度模型
• 数仓设计原则：下沉、分区、扩展
• 数据开发实施：业务代码、任务调度
• 数仓规范建设：表命名规范、字段命名、代码开发规范、任务调度规范
  
  

以上是数仓建设的一个通用思路，如果要在前面加上两个字“实时”，那么我们需要关注几点要求：

• 数据接入要即时
• ETL需要实时
  
• 仓库层数要合适
  
• 仓库任务要准时
• 数据服务要随时
  
  

实时业务架构图

原始层 ods

从上图可以看出，实时体系我们加入了对业务库的变更日志 Binlog 的处理，Binlog 日志在原始层为库级别或者实例级别，即：一个库或者实例的变更日志存放在同一个 Kafka Topic 中。事实上，APP SDK采集的数据也都接入到了实时体系。

明细层 dwd

原始层数据收集至kafka，经过清洗、转换后装载至明细层。其中对 Binlog 日志的处理主要是完成库或者实例日志到表日志的拆分，对APP SDK日志主要是做一些通用 ETL 处理。

汇总层 dws

对明细表进行加工，考虑到实时数仓对时效性要求高，不建议有太复杂的数据逻辑，一般是一个业务对应一张表，主要以宽表形式存在。

报表层 rpt

报表层是由明细层或者明细汇总层通过聚合计算得到，这一层产出了绝大部分的实时数仓指标。在汇总层的基础上对用户行为宽表的要求更进一步强化了，同时要兼顾维度的丰富性。

应用层

应用层主要是使用汇总/报表层数据以满足业务需求。应用层主要分三块：1.通过直接读取指标汇总数据做实时可视化，满足固化的实时报表需求；2. 依赖用户画像、推荐算法等数据做实时推荐；3. 实时多维即席分析需求。

总结：

数据流较短，层次相对简单，以保证数据结果的及时产出

较为单一的业务依赖，遇到故障时能对任务做迅速恢复

我们如何生产实时报表？

实时数仓技术分解

对比美团的到店餐饮业务的实时数仓架构（与爱奇艺的架构类似，基于kafka+flink）

几点优势：

• 业务分层，技术分层，弹性伸缩
• 基于hadoop生态，可灵活融合三方技术
• 全SQL支持
• 支持跨数据源join
• 数据口径统一
• 易维护
  
  

调度系统特性

• 仓库各层级任务实现分钟级调度
• 优于时间窗口模式，通过标签依赖解决数据延迟问题，保证任务队列有序进行
• 心跳监控调度集群资源，合理分配任务
  
  

Kudu调优实践

1. KUDU对磁盘IO性能的要求较高，可采用多盘或ssd 方案。

2. Master建议最多3-5个节点，Tablet Server 最多建议100-300个节点，单个节点最大8-10TB，tablets个数最大1000-4000个，单个tablet大小10-50GB，建议节点间最小带宽 10 Gbps，等待最大时长 20毫秒。

3. 表结构优化表：副本数最多7个，创建后不可更改，每张表的tablets个数在单个节点上最多不超过60个主键：不支持DOUBLE、FLOAT或BOOL类型的列，不支持自动生成的主键，最长16KB字段：不支持CHAR、VARCHAR、DATE和ARRAY等复杂类型，最多300个字段分区：支持hash、range分区，需提前创建

4. 内存优化memory_limit_hard_bytes 设置系统最大可使用内存memory_limit_soft_percentage=80% 超过该阈值，TServer拒绝写入memory_pressure_percentage=60% 进程在刷新内存中数据成为优先级之前可能消耗的硬内存限制的百分比

block_cache_capacity_mb=512M 分配给TServer块缓存的最大内存量，虽然更高的值有助于提高读写性能，但不要将块缓存容量提高到高于内存压力阈值的水平，因为这将导致Kudu在写吞吐量较低的情况下进行大量刷新，建议将块缓存容量保持在内存压力阈值（memory_pressure_percentage）的50%以下，即不超过最大可用内存的30%

举个当前集群配置的例子：32c/64g/4essd

• 
  

--tablet_history_max_age_sec=86400 # 文件回收周期，官方默认7天--memory_limit_hard_bytes=53687091200--maintenance_manager_num_threads=2--block_cache_capacity_mb=8192—log_segment_size_mb=8 # 调整文件段大小，官方默认8M

5. Skip scan optimization 原理

• 
  

SELECT clusterid FROM metrics WHERE tstamp = 100;