简介当业务发展到一定规模时，实时数仓是必不可少的基础服务。从数据驱动的角度来看，多维实时数据分析系统的重要性不言而喻。但当数据量巨大时，以腾讯为例，一天上报的数据量达到万亿级规模，要实现极低时延的实时计算和亚秒级多线程，在技术上具有挑战性。维实时查询。本文将介绍腾讯看点在信息流背景下的实时数仓和多维实时数据分析系统的技术架构。1、可解决的痛点大家可以先看看多维实时数据分析系统能解决哪些痛点。比如：推荐一个同学10分钟前发了一个推荐攻略，想知道推荐在不同人群中的效果如何？运营同学想知道，在广东省内的用户中，广东地区最流行的内容有哪些，方便做区域推送。复习同学想知道，最近5分钟游戏类被举报最多的内容和账号有哪些？老板可能想知道过去10分钟有多少用户消费了内容，对消费群体有一个宏观的了解。2.研究在开发之前，我们做了这些研究。1、线下数据分析平台能不能满足这些需求，结论是不能。离线数据分析平台不起作用的原因有：C端上报的数据需要经过Spark的多层离线计算，最终结果发送到Mysql或ES进行离线分析平台查询。这个过程至少延迟3-6个小时。目前比较常见的是提供隔日查询，所以很多对实时性要求高的业务场景无法满足；另一个问题是腾讯看到的数据量。如果太大，不稳定性会比较大，经常会出现意想不到的延迟。因此，离线分析平台无法满足很多需求。2、针对实时数据分析平台，业务组提供准实时数据查询功能。底层技术采用Kudu+Impala。Impala虽然是MPP架构的大数据计算引擎，但是它也接入了Kudu，Kudu是按列存储数据的。但是对于实时数据分析场景，查询响应速度和数据延迟还是比较高的。DAU实时查询至少需要几分钟才能返回结果，无法提供良好的交互用户体验。所以通用的大数据处理框架（Kudu+Impala）相比离线分析框架（Spark+Hdfs）更有速度优势，对于我们实时性要求比较高的场景来说是无法满足的。的。3.项目背景经过刚才的介绍，我们来看看我们项目的背景。作者发布的内容由内容中心导入，内容审核链接后启用或移除。使能的内容交给推荐系统和运营系统，再由推荐系统和运营系统在C端分发内容。内容分发到C端用户后，用户会产生各种行为，如曝光、点击、举报等，通过埋点举报实时接入消息队列。接下来我们做了两个部分的工作，也就是图中有颜色的两个部分。第一部分搭建腾讯看点的实时数仓；第二部分开发基于OLAP存储引擎的多维实时数据分析系统。为什么要建实时数仓，因为原始上报的数据量非常大，一天上报的峰值有万亿级。而且报告格式混乱。缺少内容维度信息和用户画像信息，下游无法直接使用。我们提供的实时数仓是基于腾讯看点信息流的业务场景，进行内容维度的关联，用户画像的关联，以及各种粒度的聚合。下游可以很方便的使用实时数据。4.方案选择下面我们来看一下我们的多维度实时数据分析系统的方案选择。在选型上，我们对比了业界领先的解决方案，选择了最适合我们业务场景的解决方案。第一部分是实时数仓的选择。我们选择了业界相对成熟的Lambda架构。其优点是灵活性高、容错性高、成熟度高、迁移成本低；缺点是实时和离线数据使用两套代码，可能会出现一个口径修改了，另一个口径没改的问题。我们每天都会做数据对账，如果有异常就会报警。第二部分是实时计算引擎的选择，因为Flink本来就是为流处理设计的，而SparkStreaming严格来说是微批处理，现在Strom用的不多了。考虑到Flink具有Exactly-once精度、轻量级Checkpoint容错机制、低延迟、高吞吐、高易用性等特点，我们选择Flink作为实时计算引擎。第三块是实时存储引擎。我们的需求是有维度的索引，支持高并发、预聚合、高性能的实时多维OLAP查询。可见Hbase、Tdsql、ES都不能满足要求。德鲁伊有一个缺点。它按时间序列划分段，不能在同一个段上存储相同的内容。计算全局TopN只能是一个近似值，所以我们选择ClickHouse这个近两年比较火的MPP数据库引擎。五、设计目标和设计难点我们的多维实时数据分析系统分为三个模块：实时计算引擎；实时存储引擎；应用层。难点主要在前两个模块：实时计算引擎和实时存储引擎。如何实时访问千万级数据/s，并进行极低延迟的维表关联；实时存储引擎如何支持高并发写入、高可用分布、高性能索引查询是比较难的。这些模块的具体实现，看一下我们系统的架构设计。6.架构设计前端使用开源组件AntDesign，使用Nginx服务器，部署静态页面，反向代理浏览器请求到后台服务器。后台服务基于腾讯自研的RPC后台服务框架编写，会进行一些二级缓存。实时数仓部分分为接入层、实时计算层和实时数仓存储层。接入层主要是从千万级/s的原始消息队列中拆分出不同行为数据的微队列。比如拆分后，数据只有百万级/s；实时计算层主要负责将多行行为流数据转化为行列，实时关联用户画像数据和内容维度数据；实时数仓存储层主要是设计一个适合业务的、下游方便使用的实时消息队列。我们暂时提供两个消息队列作为实时数仓的两层。一层DWM层是内容ID-用户ID粒度的聚合，即一条数据包括内容ID-用户ID，以及B端内容数据、C端用户数据和用户画像数据；另一层是DWS层，是ContentID粒度的聚合，一条数据包括ContentID、B端数据和C端数据。可以看到，Content-ID-User-ID粒度的消息队列流量进一步降低到10万级/s，Content-ID粒度更低，10000级/s，格式更清晰，维度更清晰信息更加丰富。实时存储部分分为实时写入层、OLAP存储层和后台接口层。实时写入层主要负责Hash路由写入数据；OLAP存储层采用MPP存储引擎，设计符合业务的索引和物化视图，高效存储海量数据；后台接口层提供高效的多维实时查询接口。7、实时计算本系统最复杂的两个部分是实时计算和实时存储。先介绍一下实时计算部分：分为实时关联和实时数仓。1、实时高性能维表关联实时维表关联难度大。每秒百万级的实时数据流，如果直接关联HBase1分钟的数据，需要数小时才能完成与HBase的关联，会造成严重的数据延迟。我们提出了几种解决方案：第一种是在Flink的实时计算中，按照1分钟进行窗口聚合，将窗口中的多行行为数据转化为一行多行的数据格式列。经过这一步，原来的小时级关联耗时已经降到了十多分钟，但还是不够。二是在访问HBase内容之前先设置一层Redis缓存，因为1000条数据访问HBase是秒级的，而访问Redis是毫秒级的。访问Redis的速度基本上是访问HBase的1000倍。为了防止过期数据浪费缓存，缓存过期时间设置为24小时，通过监听写HBaseProxy来保证缓存的一致性。这会将访问时间从十分钟更改为几秒钟。三是在举报过程中会举报很多不规范的内容ID。这些内容ID并没有存储在内容HBase中，会造成缓存穿透的问题。所以在实时计算的时候，我们直接过滤掉这些内容ID，防止缓存穿透，减少一些时间。第四是因为时序缓存，会引入缓存雪崩问题。为了防止雪崩，我们在实时计算时进行了削峰填谷操作，并错开设置缓存的时间。可以看到，优化前后，数据量从百亿减少到亿，耗时从几小时减少到几十秒，减少了99%。2、下游提供实时数仓服务的难点在于属于比较新的领域，各个公司的业务差距比较大。如何设计一个方便易用、满足有趣业务场景的实时数仓是难点。我们先来看看实时数仓是干什么的。实时数仓对外有多个消息队列。不同的消息队列存储不同聚合粒度的实时数据，包括内容ID、用户ID、C端行为数据、B端内容。维度数据和用户画像数据等。我们如何构建实时数仓，就是上面描述的实时计算引擎的输出，存储在消息队列中，可以提供给多个下游用户复用。我们可以看到在构建实时数据仓库之前和之后开发实时应用程序之间的区别。在没有数仓的情况下，我们需要每秒消费数千万级的原始队列，进行复杂的数据清洗，然后进行用户画像关联和内容维度关联，获取满足需要的格式、开发和扩展的实时数据成本会比较高。如果你想开发一个新的应用程序，你必须重新经历这个过程。有了数据仓库之后，如果要开发ContentID粒度的实时应用，可以直接申请DWS层的消息队列，TPS为10000/s。开发成本低很多，资源消耗小很多，扩展性强很多。让我们看一个实际的例子。为了开发我们系统的实时数据大屏，我们需要执行以上所有操作来获取数据。现在只需要消费DWS层的消息队列，写一个FlinkSQL，只需要消耗2个CPU核和1G内存。可以看出，以50个消费者为例，在建立实时数仓前后，下游开发实时应用可以减少98%的资源消耗。包括计算资源、存储资源、人力成本和开发者学习接入成本等。而且消费者越多，节省的越多。以Redis存储部分为例，一个月可以节省上百万RMB。8、实时存储介绍完实时计算，我们再介绍一下实时存储。本节分为三个部分进行介绍：分布式——高可用；海量数据——写入；高性能-查询。1、分布式-高可用这里听听的是Clickhouse官方的建议，使用ZK实现高可用的方案。数据写入一个shard时，只写入一个副本，然后写入ZK，ZK用来告诉同一个shard的其他副本，其他副本来拉取数据，保证数据的一致性。这里没有使用消息队列进行数据同步，因为ZK比较轻量。并且在写入的时候，任意写入一个副本，其他副本可以通过ZK获得一致的数据。并且即使其他节点第一次获取数据失败，只要发现与ZK上记录的数据不一致，就会再次尝试获取数据，保证一致性。2.海量数据写入数据写入遇到的第一个问题是如果直接将海量数据写入Clickhouse，ZK的QPS会过高。解决办法是使用Batch模式写入。批量设置有多大？batch太小无法缓解ZK的压力，batch也不能太大，否则上游内存压力太大。通过实验，我们最终选择了一个大小为几十万的batch。第二个问题是，随着数据量的增长，每天可能有数百亿的数据写入单个QQ手表的视频内容。默认方案是写分布式表，会导致单机出现磁盘故障。瓶颈，尤其是Clickhouse的底层是Mergetree，类似于HBase和RocketsDb的底层LSM-Tree。在合并的过程中，会出现写放大的问题，增加磁盘的压力。峰值是每分钟几千万条数据，需要几十秒才能写完。如果在做Merge，写请求会被阻塞，查询会很慢。我们有两种优化方案：一种是对磁盘进行Raid，提高磁盘的IO；另一种是写入前先分表，直接分别写入不同的shard，磁盘压力直接变成1/N。第三个问题，虽然我们把写入分片了，但是这引入了一个分布式系统的通病，就是局部的Top不是全局的Top。比如同一个contentID的数据落在不同的shard上，计算全球Top100读取的contentID。一个contentID在shard1上是Top100，但是在其他shard上不是Top100，汇总的时候会丢失一部分数据，影响最终结果。我们做的优化是在写入前加了一层路由，将相同内容ID的所有记录路由到同一个分片，解决了这个问题。介绍完写，接下来就是介绍Clickhouse的高性能存储和查询。3.高性能-存储-查询Clickhouse高性能查询的一个关键点是稀疏索引。稀疏索引的设计很有讲究。好的设计可以加快查询速度，但不好的设计会影响查询效率。基于我们的业务场景，因为我们查询的大部分都是跟时间和内容ID相关的，比如某个内容在过去N分钟里在各个组里的表现如何？我有按日期、分钟粒度时间和内容ID排列的稀疏索引。对于某个内容查询，建立稀疏索引后，可以减少99%的文件扫描。另一个问题是我们现在有太多的数据和太多的维度。以QQ观看的视频内容为例。每天有数百亿条流，某些维度上有数百个类别。如果一次预聚合所有维度，数据量会呈指数级增长，查询反而变慢，占用大量内存空间。我们的优化，针对不同的维度，建立对应的预聚合视图，以空间换取时间，可以缩短查询时间。分布式表查询还有一个问题。查询单个内容ID的信息，分布式表会将查询发送到所有分片，然后返回查询结果进行汇总。事实上，由于路由的原因，一个内容ID只存在于一个分片上，其余分片都空着运行。对于这类查询，我们的优化是在后台按照相同的规则进行路由，直接查询目标分片，减少了N-1/N的负载，大大缩短了查询时间。并且因为我们提供的是OLAP查询，数据只需要满足最终一致性，通过主从副本读写分离可以进一步提升性能。我们还在后台做了1分钟的数据缓存。同样的查询，后台会直接返回。4.扩容这里我们再介绍一下我们的扩容计划，并调研一些业界通用的解决方案。例如在HBase中，原始数据存储在HDFS上。本次扩容只是RegionServer的扩容，不涉及原有数据的迁移。但是Clickhouse的每一个shard数据都是本地的，是一个比较底层的存储引擎，不能像HBase那样容易扩展。Redis是一个类似于consistenthashing的hashslot，是比较经典的分布式缓存方案。虽然在rehash过程中Redisslot暂时无法用于askreading，但整体迁移还是比较方便的。从原来的h[0]迁移到h[1]，最后删除h[0]。但是Clickhouse大多采用OLAP批量查询而不是抽查，而且由于列式存储不支持删除，一致性哈希方案不是很合适。目前的扩容计划是再消费一条数据，写入新的Clickhouse集群，两个集群一起运行一段时间，因为实时数据会保存3天。3天后，后台服务会直接访问新集群。9.成就腾讯观点实时数仓：DWM层和DWS层，数据延迟1分钟。Foresight多维实时数据分析系统：亚秒级响应多维条件查询请求。在缓存未命中的情况下，过去30分钟内99%的查询耗时不到1秒；过去24小时90%的查询耗时在5秒以内，99%的请求在10秒以内。
