服务 15 亿 + 游戏用户数据的腾讯 iData 分析中心进化之路:计算引擎基础能力构建

阅读数:56 2019 年 11 月 5 日 16:18

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

腾讯游戏分析 iData 是国内领先的智能化数据服务平台,致力于为游戏提供一站式数据分析,上线至今覆盖了 567+ 款业务,涵盖 15 亿以上的游戏用户数据。精炼游戏多年经验核心指标,60 万个以上的可视化图表服务加专业化游戏多维分析,基于用户生命周期提供完整的数据化方案,实现数据驱动的精细化运营,对公司游戏运营提供了完善的数据化运营工具,帮助游戏更全面,准确,实时的分析数据。

腾讯 iData 分析中心是 iData 产品的重要组成部分之一,负责号码包提取、画像分析、工作分析等围绕用户号码包的数据分析功能。在长达几年的运营之后,针对运营中产生的一些问题和用户的新需求,我们意识到了旧系统的不足,开始打造新的分析中心后台。我们将以系列文章,围绕新分析中心后台 TGMars 的计算平台的方方面面,来探讨、介绍我们是如何思考、研发新分析中心的。

上篇文章《为什么要用 Spark?》中,我们探讨了在计算平台上的选型,为什么选择基于 Spark 构建计算平台,我们是如何使用 Spark 的,以及基于此又做了什么更多的工作。本文将探讨 TGSpark 如何通过 Spark 提供的扩展接口使得 Spark 可以加载 TGMars 上的数据,这将作为 TGMars 的计算引擎基础,我们会基于这套机制深入定制我们的需求,形成独有的 TGMars 计算能力。

一、为什么需要自定义的数据源?

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

很多计算引擎都有自定义数据源的能力;显然,即使各种大数据计算平台都已经默认支持 HDFS/S3/JDBC 等,但是不能排除这样一种情况:用户有自定义的数据需要和计算平台对接,或是旧系统的兼容,又可能是为了特定场景的设计,TGMars 正是后一种情况——我们需要为一种特定的场景优化存储。

TGMars 存储有什么特点?

具体的可见文章《新一代 tGmars 存储架构浅析及场景测试》,TGMars 存储是延续之前 iData 分析中心中位图存储进化而来的,除了该文章提及的,TGMars 存储解决了各种之前系统中缺失的数据容灾和数据迁移功能以外,还延续了之前设计的优点:

  • 维护同业务下相同的数据表分片规则
  • 维护数据文件的上传与分发
  • 分片数据均匀分发到存储节点上
  • 保障相同分片数据位于同一组机器

同一业务分片规则固定,且相同分片数据在同一机器,这使得计算节点加载数据时,相同业务(数据库)处理同一 id 的分片时,都可在同一个任务节点上计算完成,尽量不让数据出现 shuffle。

这也是延续之前我们在位图分析系统中系统的优势时提到的,是值得保留的特性,既然如此,在 TGSpark 中,我们必须也要让 Spark 能够适配这种特性,使得计算节点跟随存储分发,保证数据在本地读取与计算。

Spark 配套的默认存储则是 HDFS,它的存储逻辑则很不一样:文件按等长分块后分布到存储节点中,并不是数据预分片的。它也有自己的好处:计算部分不必过于关心数据节点本身,可以选择从网络直接加载;但同时又可以提供分块的节点信息,适配某些分布式计算系统的数据并行加载,是一种兼容性很高的设计。正是这样 HDFS 是很多大数据方案的首选存储,比如 CarbonData 虽然是自定义数据,也同样适配 Spark 计算,但是存储仍然是 HDFS,这样数据仍然是分块并行加载的。

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

不过既然我们确定了数据固定分片的原则,HDFS 自然是不可能采用的,这也是为什么要自定义数据源的缘故。

二、Spark 的 DataSource API V2

既然是用 DataSourceAPIV2 来自定义数据源,那么就有 DataSourceAPIV1 咯?

其实对于 Spark 来说,更早的是 CustomRDD 机制,比如 Spark 自带的 HadoopRDD 和 JdbcRDD,这是基于 RDD 的做法,在 1.x 的时代,还在大规模使用 RDD 的时候的首选。

不过在 2.x 时代,Spark 实际上渐渐雪藏了 RDD,而仅仅是作为底层的一种存储结构,DataFrame 进化成了 Dataset,并大规模重用:MLlib 已经被废弃,新的 ML 库基于 Dataset,取代了旧 MLlib;Dataset 支持以前的类 RDD 操作,照旧可以操作更低层的算子;Dataset成为了新的 DataFrame,支持 SQL 操作,还有 Catalyst 和 Tungsten 的提携,性能提升十倍;相比旧 RDD 官方寥寥仅支持的 JDBC(难用还受不推荐)和 Hadoop 载入方式,Dataset 还支持多种数据源,使得数据的读取更加灵活。

官方爱 Dataset,甚至将 DataSourceAPI 赐给 Dataset,叫用它的人不仅仅能加载 HDFS 文件或 Hive 表,反能通过 DataSourceAPI 加载更多的数据!

不过 DataSourceAPI 也有自己的缺陷:

  • 依赖 SQLContext,作为一个数据源 API 却依赖了高级接口,设计不合理;

  • 不支持列存常见的列选取,如果是列式存储则需要侵入性的使用内部接口;

  • 分片数据加载时候 Executor 和对应数据存放的位置不能很好的保证一致;

  • 写数据不支持事务,不能通知被写入的数据源清理失败的任务;

  • 数据源的扩展性较差。

因此现在官方推荐了 DataSourceAPIV2,虽然还是个不稳定的演进接口,但是整体的设计原则却较为固定,仍然可以一试。

三、SparkDataSourceAPIV2 的一些特性

DataSourceAPIV2 扩展性很好,其几个接口保证了它可以按照我们预想的方式加载到数据:

1. SupportsPushDownRequiredColumns

下推选列到数据源,使得数据源只提供必要的数据

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

2. SupportsPushDownFilters

下推过滤条件到数据源,使得数据源提前过滤部分数据

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

3. SupportsScanColumnarBatch

提供按列方式的数据加载,无需再把列式存储的数据转置为行

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

4. SupportsReportPartitioning

报告数据分片规则,让 Spark 进一步取消不必要的 Shuffle

5. InputPartition.preferredLocations

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

报告分片所在 ip,使得下发数据可以尽可能本地加载。

这样看下来,我们设立的原则,Spark 已经都考虑到了,1、2 保证选列和下推的实现,虽然旧 API 也有,但是 V2 的 API 更好用;3 则考虑到了流行的列存方式;4/5 则是其中的重点,保证本地计算的关键。

这其实得益于 Spark 一开始就存在的 Catalyst 优化器机制。SparkSQL 会根据数据的性质优化查询计划和物理计划。

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

可以简要的了解下 SparkSQL 是如何执行 SQL 语句的:

1.SQL 语句初步翻译为一个未绑定到数据源的逻辑计划,此时查询计划还没有绑定数据源

2. 通过 Catalog 机制,查询计划被绑定到特定的库表和列上,进一步检查查询是否能够执行,形成逻辑计划

3. 逻辑计划进过优化器优化,形成优化的查询计划

4. 查询计划转变为一系列物理计划

5. 物理计划通过对应的执行策略,转为实际动作,加载为 RDD(Dataset)

选列、下推和分片在生成物理计划前的执行策略中得以优化,支持选列、下推操作、分片报告时,扫表操作将只有选中的列输出,支持的下推则取消,分片规则符合条件的将去掉一层重分区 /Shuffle 操作。Spark 具体是怎么实现的?那么等我介绍扩展点注入的时候再说。

本地加载机制则在物理计划执行时生效,它绑定到 RDD 的 preferredLocatin,在 Spark 内部的 TaskScheduler 生成任务时尽可能的去绑定到指定的机器上。

四、来做一些实验吧

写这篇文章的时候,基于 DataSourceAPIV2 的 TGMarsDataSource 已经开发的差不多了,因此我们可以来做一些实验验证下,我们提出的几个关键点是否已经都能实现。

我们这里会基于示例的表来看特定 SQL 下的查询计划来验证。

基础表描述如下:

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

首先看看列选择:

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

在这个例子里可以看到最终调用 ScanV2 任务扫表时,虽然优化的逻辑计划里面还有全表扫描操作,但是到最后的物理计划时,ScanV2 任务变成对应 Project 任务需要的,只对应列信息。

然后是过滤下推:

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

在这个例子里,where 条件包含两个条件 platid=0 和 level=10,这触发了过滤下推的支持,到最后的物理计划时,ScanV2 任务会带上两个 Filter,同时不再上层做多余的 Filter 操作。

最后再看一下分片规则的应用:

服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建
服务15亿+游戏用户数据的腾讯iData分析中心进化之路:计算引擎基础能力构建

因为我们会根据主键列分片,因此输出包含主键的时候,distinct/group by 时可以跳过一层多余的 Exchange 操作,这个时候计算是数据和计算都在本地进行的,没有 Shuffle,而如果只有其他列,则不能实现本地计算, 必须 Shuffle。这种方式使得我们可以同时兼容以前的计算方式,还能藉由 Spark 的能力支持更灵活的计算。

五、留个伏笔

我们借由 Spark 的 DataSource API V2 的实现我们的诸多目标。不过我们在这里再讨论下本地计算的问题:

如果我们支持本地计算,那么我们必须保证一个原则,有多少存储节点,对应下发多少 Executor,而且我们必须要保证 Executor 均匀的分布在各个节点上。这里我们是否可以确保这点呢?

实际上考虑这样的一个情况,我们有 8 台机器,但是由于任务资源的紧张,我们只有 4 个 Executor 可用,那么此时就有 4 个计算节点对应 8 个存储节点,这样必然有些数据不是本地加载的。

还有一种情况,如果我们有 8 台机器,但是只有 4 台机器有数据,那么又怎么保证计算节点准确的分发到了存储节点呢?

这个需要深入研究 Spark 的 Executor 分发机制,再考虑我们如何应对这个问题。

本文转载自公众号云加社区(ID:QcloudCommunity)。

原文链接:

https://mp.weixin.qq.com/s/caEOJ4V7Rs1Nc1MUggdPFg

评论

发布