吴瑞诚分享---斗鱼大数据的玩法

编者按：在大数据学习交流付费群里，斗鱼数据平台部总监吴瑞城分享了斗鱼大数据的玩法，以下是分享内容。

我是吴瑞诚，现在负责斗鱼数据平台部，今天给大家分享一下斗鱼大数据这块的玩法。我先做个自我介绍，我是 11 年初华科通信硕士毕业就进入淘宝，主要做 HBase 相关开发，后来回武汉后在 1 号店转向应用架构方向。

我是 14 年 9 月加入斗鱼，当时斗鱼研发是 30 人的规模，从 0 开始搭建斗鱼大数据平台，单枪匹马一个人，大概干了三个月，招不到大数据开发，哪怕只是基本了解 Hadoop 的都很招不到，干的很苦。所以团队第一批组员是组内培养的，会一些 Java 开发基础的应届生，生拉硬拽凑到大数据团队。其中一位武汉理工 15 年的本科应届生，现在已经成长为我们部门 BI/DW 团队的 Leader。到现在，加上即将入职的应届生，数据平台部团队规模接近 60 人，分为四个组：大数据处理组、BI/DW 组、数据挖掘组、基础架构组。

在武汉招开发难，招大数据开发更难，这也是我、柴楹和猫友会发起这个收费群的主要目的。让整个武汉互联网圈看着起势的时候，会有一个大数据的圈子。大家多沟通，多交流，把整个圈子的气氛带动起来。这样对于我们研发人员本身和武汉本地的公司都是极大的利好。

先简单介绍一下斗鱼：1.国内最大游戏直播平台，从 16 年开始向泛娱乐化方向发展； 2.日活用户 2000 万，最高同时在线人数约 500 万； 3.主播日活 40,000 人，每天产生原创视频数万小时。； 4.ALEXA 排名：全球约 200 名、国内约 20 名（近期略下滑，今天中午看在 30 多名，比优酷、Bilibili 等站要高）；

这是一个典型的斗鱼直播间，是斗鱼最主要的内容形态：左边是视频区，上面飘过的文字就是弹（ dàn ）幕，直播网站上用户和主播互动的最主要形式。弹幕服务是性能压力最大的服务之一，相当于是需要百万人群聊时的消息推送量，后续有机会和大家分享；视频区下面是礼物赠送区；右侧是弹幕区，Tab 页上是排行榜，用户对该直播间的贡献值，土豪喜闻乐见；

接下来我会跟大家分享一下斗鱼大数据现在的玩法和下一步的规划。主要内容分为四方面： 1.斗鱼大数据平台的整体架构； 2.斗鱼数据仓库 3.斗鱼个性推荐系统 4.斗鱼风控系统

1、斗鱼大数据平台整体架构

先来看看斗鱼大数据平台的整体架构，上一张镇楼的图：

14 年加入斗鱼，当时负责的第一个业务是用 shell 脚本，从主站服务器上拉 Nginx log 到 Hive 集群中，统计产生报表。随着网站数据量的增长和配备约来越完备，从最开始的小系统越做越大，做到现在这个架构。

这个架构差不多是在 16 年初的时候成型的，主要包括：数据源层，用户行为打点、服务日志、Nginx/PHP 日志（离线和实时两条数据流）、线上 MySQL 库、MongoDB、Redis 存储数据接入层，包括了 Kafka，负责接入打点上报（最大吞吐量约 200w TPS ）、Canal/Sqoop （主要解决数据库数据接入问题）、Flume/Logstash/rsyslog 等实时日志；数据预处理层，简单的清洗和聚合计算；存储层，典型的系统 HDFS （文件存储）、HBase （ KV 存储）、Kafka （消息缓存）；计算层，主要包含了典型的所有计算模型的计算引擎，包含了 MR、Hive、Storm、Spark 以及深度学习平台比如 Tensorflow 等等再往上就是数据服务层，主要提供对外数据服务。这一块现在主要是基于 Docker 实现的一整套微服务环境来支撑；

1.1 大数据管理平台

下面先简单介绍一下各个组件的玩法，首先是大数据管理平台，承载所有的元数据管理、统一监控系统、报表展示、任务调度、发布系统等所有配套功能。

1.2 基础集群架构

可以看到，清晰的区别了 online 和 offline 集群 机器配置： ● 现负责约 300 台高配服务器，Dell 730xd，32Cores 128/256G 内存，16*6T Sata Disk ； ● Hadoop 大集群 150 台物理机，5PB 数据，日增量约 20T ；

实时离线集群分离，避免资源竞争造成业务处理抖动。集群分离后也可以更安全的对集群进行运维操作； 使用 Kafka 作为 MQ，更确切说是消息通道；

1.3 OLAP 查询引擎 Impala

Impala 之前，我们有使用 Presto 的经验，后来因为运维上的问题，暂时下线了。Kylin15 年很早的时候，在北京和 kylingence 官方有过一次深入沟通，仔细评估后发现斗鱼的场景不能发挥他的优势。最近 Kylin 重大版本之后，准备再评估一次。这方面应该有不少同学都有相关的经验，最后讨论阶段，大家可以一起聊一下。

可能对大数据组件不太熟悉的同学，可能对 OLAP 概念不太熟悉，主要是针对秒级大数据量查询场景。对应的 OLTP 是针对事务处理，我们比较常见的 MySQL 和 Oracle 属于这类。这样解释大家可能接受起来会简单一些。

使用 Impala 作为 SQL 查询引擎，相比 Hive 整体提升 5 倍速度；使用了完全区别于 MapReduce 的一套数据处理模式。

我们主要做了一点：由于 Impala 是 C++编写，为了提升性能，将部分高查询密度的 UDF （如：JSON 解析）替换为 C++实现。使用 HA Proxy 作为 Impala 负载均衡器，均衡 JDBC 连接。避免 Impala Daemon 故障、重启时影响正常业务。这样，可以覆盖我们现在大部分的秒级大数据量查询（ 10 亿级别）

1.4 Spark 应用

Spark 在斗鱼大数据生态系统有着举足轻重的作用。涉及到： ● 接入实时流（ 7*24h ） 各个场景的实时 PV、UV 统计，实时日志分析，服务接口性能监控，用户分布及其他报表等。已上线独立实时任务 20 余个。 ● 数据抽取与处理 抽取异构数据源（ MySQL，MongoDB ），将其抽取数据合并、加工后写入数据仓库。每日凌晨触发数据抽取加工任务。为最重要依赖调度任务，为后续每日离线任务准备数据。 ● 数据分析 /挖掘 包含弹幕统计分析，流量渠道统计分析，垃圾弹幕识别，用户点击预测，用户分群，推荐系统等任务。这也是部门今年非常重视的一块，任务规模也在日益庞大。

部署：Spark on Yarn，单节点部署，依赖 Yarn 环境 发布：基于自研发布系统，可配置一件部署 调度：基于自研发布系统，定时调度与依赖调度两种模式可选 告警：监控与告警系统支持任务执行监控及其结果校验监控

由于正在经历 Spark 大版本升级（ 1.6.x2.1.x ）进程之中 ，斗鱼数据平台部线上存在两个 Spark 集群来支撑目前的业务 ● 1.6.3 版本 Spark 基于线上 yarn 进行部署（ Spark on Yarn ）：66 个节点 72G 内存 /节点 22 个核 /节点 ● 2.1.0 版本 Spark 基于 Standalone：12 个节点 72G 内存 /节点 22 个核 /节点

Spark 生态中，我们主要用到的两个组件： 1.Zeppelin 主要提供 Spark SQL 交互界面查询，数据可视化支持。后续将支持 Python、R 语言的建模，支持用户 Notebook 的提交和调度。 2.Alluxio 基于内存的分布式文件系统，它是架构在底层分布式文件系统和上层分布式计算框架之间的一个中间件，主要职责是以文件形式在内存或其它存储设施中提供数据的存取服务。

底层的存储层和计算层的基础集群大概是这些。

1.5、基于 ELK 的统一日志监控系统

现在我们把 es 集群按业务场景划分为多个小集群，这样的可以避免不同业务出现抢占资源的情况。

多个 ES 集群，50+物理节点、每日 15T+日志量，多实例部署，接入全站所有服务器日志； 年后，完成全部升级至 ElasticSearch 5.X，性能提升明显； 废弃 FlumeAgent，使用 Firebeat、Rsyslog，小巧稳定、资源占用低； 为部分业务独立开发日志解析器，提高性能，有 Java 版和 Spark 版；

我们应用 ELK 的场景中，最难的就是在 agent 资源占用和 agent 抽取吞吐之间做权衡，想马儿快，又想马儿不吃草，会针对不同的语言栈，使用不同的 agent 是实现。

推荐日志使用 JSON 格式，降低解析压力、增减字段灵活；

我们 agent 会把日志灌入 kafka，然后从 kafka 出口的日志流就需要稳定、格式化的结构往 ES 集群小 batch 灌，这样就出现了使用 hangout 和自研数据管道两种实现方式，hangout 是 java 版的 Logstash，效率仍然满足不了我们的要求，所以，有了基于 spark 的自研日志消费管道。

对 ELK 的使用，算是踩坑无数。讲一些 ES 主要优化点： 1.索引按小时切分，当索引无数据写入时进行 ForceMerge 提升查询性能； 2.由于日志收集写请求远远大于读请求，保证每个节点有分片分摊写压力（每个节点 1~2 个分片），节点磁盘独立； 3.每台服务器（ 128G 内存），1Master+2Data，预留大约一半内存作为 System Cache 提升查询性能；记住，一定要留足 cache。。 4.cluster.routing.allocation.same_shard.host 禁止主从分片被分到同一台服务器上（不同节点），保证服务器宕机时索引可用；

ELK 的整体优化思路是为了能抗住低延迟和大日志量的问题；现在我们已经能稳定在每日 15T+日志量级，基于这一套统一日志监控系统，做了很多业务、服务的监控和告警。

我们现在吞吐量最大的实时数据流，100w+TPS，全站各个端所有用户行为的实时监控，可以实时看到整体打点水位、各个客户端、各个版本播放器的健康状况

这是推荐接口的实时 Dashboard 监控效果图：左上是每次请求的最大耗时、右上 50 分位和 99 分位耗时统计；左下是推荐接口的整体水位统计、右下是各个 Tomcat 实例的水位统计；这样可以对实时推荐接口的整体访问量、每个实例的请求量、性能水位、超时请求一目了然，监控神器。基于此可以做阀值的监控，解放双手。

现在每个核心接口分给不同同学来负责，人手配置一个类似的 Dashboard。加上告警，就可以腾出手来了。

再分享最前面有一个大的架构图，最上层是数据应用层，我们会对外提供多种服务，包括个性推荐、实时监控、广告系统、风控系统、搜索引擎（基于 ES ）、后台数据应用等服务都是由我们自己来完成整个服务的部署，用以对外提供访问；这样，除了数据层面的处理，也对我们工程实现上提出了高要求，要能直接提供对外服务。我们现在的玩法是基于 Docker 生态构建完备的微服务体系来实现，直接上图

前后端完全分离，使用 Nginx 作为网关，代理后端服务。功能较单一，正在预研其他网关方案； 全面升级至 SpringBoot，拥抱微服务；

1.使用内嵌的 Web 容器（ Tomcat、Jetty、Undertow ），由应用自己掌控整个生命周期，形成闭环； 2.容器化改造更平滑，提高应用弹性（方便统一底层系统环境、方便扩缩容）； 3.内置的 Metrics 接口使应用监控更方便。

这样，就讲到斗鱼服务容器化 1.使用 K8S 作为容器编排引擎，进行容器调度、运行、滚动升级、灰度发布； 2.暴露服务状态接口（ isReady，isHeahthy ），更好的利用 K8S 探针进行自我健康检查； 3.容器网络使用 Flannel （ VxLan 模式）； 4.使用 Nginx 作为网关代理外部请求（非容器网络内的请求）; 5.进行容器化改造的过程中，对于需要暴露接口的服务，初期可以将服务容器网络设置为宿主机网络，避免接口无法访问的问题（尤其是在服务发现场景，注册的是容器 IP，外部无法访问），后期进行宿主机端口感知、注册。

主要挑了核心组件来做简单分享，抛个砖；当然为了支撑 Devops 开发模式（每位同学从需求分析开始，一直负责设计、开发、测试、发布、运维、告警、优化等服务全周期），需要有配套的任务调度系统（基于 ZK 自研）、发布系统（基于 Jenkins 自研）、监控系统（自研）等系统限于篇幅不做展开；

2、斗鱼数据仓库

斗鱼数据仓库最开始就是一个 Hive default 库，导入的表是 Nginx/PHP log。慢慢需要对注册用户进行统计分析，逐渐导入了注册表、礼物流水、充值表等等关系表。逐渐增加到近 200 张表粗放的放在 default 库，碰到权限、误删除等问题后，数据仓库分层刻不容缓；

斗鱼数据仓库分层主要分为以下几层：ods：ods 层主要用来存储从业务数据库表，如 MySQL，MongoDB 同步过来的原始数据，以及线上用户行为等原始数据 external：external 层主要用来存放线上写到 HBase 数据建立的外部表，以及日志数据的外部表 dim：dim 层主要用来存放维度表信息，如直播间维度信息等 dwd：dwd 层主要用来存放从 ods 层以及 dim，external 处理清洗过后的数据，方便接下来的计算 dw：dw 层主要用来对用户或主播等进行一些轻维度的汇总 ads：ads 层主要是应用层的一些数据，如对外的报表数据 archive：archive 层主要是归档历史数据

数据仓库中还会根据业务以及数据类型做域的划分： 业务域：ad-广告，game-游戏数据域：log-日志域，pay-交易域

斗鱼数据仓库规范实施数据仓库基本的建设完成后，就会有一些用户操作数据的规范，如：SQL 规范，数据时间周期的规范 SQL 规范主要有以下几个方面类型统一：double，string，bigint 数据倾斜：group，count distinct，开窗函数表级&字段级注释条件语句类型必须一致 case when 语句必须要有 else 表查询带上库名称

开始同步数据时使用了开源的 sqoop 作为同步工具，由于对多数据源的支持不够丰富以及对内部系统的兼容性问题，逐渐放弃使用；

自研了 Data-Porter 工具作为目前主要的多种数据源与数据仓库间的同步工具。Data-Porter 基于 Spark Pipeline，将数据从 DB 读取为 Spark RDD，再转为 Spark DateFrame，最后使用 Spark SQL 将数据写入 Hive，这种通过 API 由低阶到高阶的应用，实现了无附加操作的数据同步 。

目前支持 MySQL 的数据库分库分表的同步，MongoDB 集群同步以及数据源去重，但是也有一些不足的地方，比如对数据源目前还只支持了 MySQL 和 MongoDB，后续会规划开发关于 HBase 等目前主要使用的数据源的同步支持。

3、斗鱼个性推荐系统

先看看斗鱼个性推荐的主要栏目位：

• App 首页几乎都是个性推荐的栏目位，每个用户看到的都是依据自己口味推荐的直播间，千人千面；
• 直播列表中有特定的两个位置；
• 直播间页面中，有“超管推荐”，每个直播间看到的都不一样，百人百面。

其中 App 首页是完全千人千面，不同的分区、不同的房间，都是根据用户历史行为，预测的用户偏好进行推荐。可以看下斗鱼个性推荐的场景，很多，而且会越来越多。是公司的总体战略目标。

斗鱼个性推荐是从 15 年开始预研，线下做推荐方案对比、模型效果对比、实际数据试跑，到 16 年 6 月份才正式上线。从 Web 端的列表、直播间内的推荐位，同步 APP 端上线。现在 APP 首页大部分都是个性推荐位，做到千人千面的实时推荐；

个性推荐的算法模型选择是一方面，推荐服务本身工程实现的性能也是很大的考验，要支持 500w 的用户同时在线量，接口处理请求峰值近 1w +QPS。得益于服务化，核心功能都配备完备的监控告警。正在上线的 Docker 容器化，可以大大提升服务的弹性，方便服务规模的扩缩。

个性推荐的效果好坏拼的就是用户画像，对自己用户越熟悉，推荐效果才可能越好；斗鱼的用户画像现在是这么玩的：

从上面的图可以看到，目前我们的推荐系统主要分为三个模块：推荐服务层、数据维护层、监控层。

1)推荐服务层推荐业务规则过滤，根据业务需求过滤不需要推荐的数据推荐算法配置，根据业务场景及推荐位配置不同的推荐算法 ABTest 分流，用于推荐算法对比及灰度上线结果数据格式化及排序 2)数据维护层基础信息数据使用 dubbo 提供服务，降低服务层对数据源的依赖基于物品相似、用户标签、KPI 体系等算法，离线计算与实时计算相结合推荐算法数据存储，基于不同算法数据分开存储维护数据缓存策略，提升缓存命中率，降低 DB 请求 3)监控层接口可用性监控、接口性能监控数据质量监控，保证数据准确性推荐效果监控，实时关注推荐转化率

个性推荐无法速成，斗鱼个性推荐当前也有不少问题： a.推荐规则单一，点击转化率约 5 成，仍有提升空间 b.推荐服务开发效率及扩展性不高 c.推荐数据分散，复用率低

4.斗鱼风控系统

这个图片有一个很励志的故事，是这样的：20 平米房间，2 万张卡，前期投入约 40 万，一天换一次，一个月内所有的卡都跑一圈。每月收入 30 万，高峰月入百万。工作室的日常：每天晚上将卡一张张取下来，更换新卡。这是黑产的一个缩影。

那斗鱼面对的黑产风险有：

这是前天我从淘宝上搜索斗鱼 tv 时的结果截图，可以作为现在斗鱼面对黑产形势的一个典型缩影。

主要集中在直播间人气（主播价值的一个衡量依据）、鱼丸（主播可兑换的礼物）、打折鱼翅（黑卡充值重灾区，主播可以兑换，危害更大更直接）、主播刷关注，可以从淘宝价格上一瞥黑市上各个刷量的难易和获益程度。同时，在搜索结果中，也能看到斗鱼友台出现在搜索结果中，可见，黑产风险是整个行业要面对的，不仅仅是斗鱼。在这一块，是需要行业内的整体合作的，尽管这个合作有些障碍。

斗鱼风控系统是这样的： 1 大力惩治的同时，要保证用户体验，尽量降低误杀； 2 依据用户行为数据获取特征模型，提取作弊行为的特征模型； 3 针对 IP 和设备画像，对 IP 和设备进行嫌疑级别评级打分； 4 采用线上实时+离线分析 双层识别模型为主，辅以“基于规则-rule-based ”判断； 5 提取用户行为风险模型、用户风险评分等级

对照上图，挨个注释一下： 建设决策中心(Drools)，识别各种作弊行为， 优化反作弊算法模型。风控实时引擎， 以实时弹幕、异地登录等关键行为实时流为基础，在结合账号防盗及充值消费数据，实时评估用户风险等级。

用户行为轨迹分析系统，协助分析用户作弊行为，提炼风控规则。风控离线引擎， 不定期更新离线风控规则，接入决策中心，统一配置。

风控 web 管理平台，用于配置决策规则、黑白名单管理、行为轨迹查询等功能。风控数据服务化，对外提供安全访问的接口，实时查询用户风险行为，降低损失。

总结： 我们有很多数据；围绕数据我们做了很多努力；要让数据发挥更大的价值

问答：

1、快速发展的小公司有哪些潜在的大数据处理场景？需要用到哪些关键的技术和架构？除了个性化推荐，还有哪些大数据落地的例子？离线计算在斗鱼的应用有哪些？

答：我个人感觉上大数据一般都会被业务推着走，如果是特意要去玩大数据，可能会比较难推动。比如我从淘宝回武汉，在一号店干了一段时间，在某些场景，确实是不合适。。所以我建议，用集群的方式，先做一个简配版数据仓库，拿到了数据，才好想一些玩法。这个问题大概是这样。

2、看你们刚刚的集群有好几个，分为计算集群，和实时等，为啥不用 yarn 的队列机制，分队列进行弹性计算，这样可以充分利用集群。cdh 有推出资源池，有没有去尝试？好几个集群维护比较麻烦，你们这帮是怎么管理的？

答：我们现在用的 cdh 版本，也是基于的 yarn 管理集群资源。

3、对于北上广漂着的想回武汉发现的程序猿们有什么建议和忠告？

答：这个问题我有经历，所以有发言权。我当时回武汉，是因为家庭原因。所以我是没有选择的情况下回的武汉。现在武汉的坑位比早两年已经多了很多，公司也多了很多。大公司的研发中心，本土起来的互联网公司都有不少。所以坑位有了，就看自身的成长，能不能满足这些坑的期望。我们的 offer 被好多公司都有抢过，所以薪资这块，我觉得也不是问题，呵呵。

4、斗鱼的负载均衡是如何做的？斗鱼数据安全是怎样保证的? 如果想构建自己的 VPN 服务器，有啥要注意的?

答：按我的理解来回答，网站入口是 DNS-lvs-nginx-服务容器，数据安全依靠权限管理（账号-角色-权限）。构建自己的 VPN，这个我就不太了解的。太偏网络层面，得找 IT 同学帮忙。

5、我在北京工作已有五年。先后在百度,小米做过游戏视频分发的开发工作。大数据这一块接触的比较少。年后想回武汉发展。请问斗鱼有合适的坑吗？

答：游戏和视频分发斗鱼现在没有自己做，但是从你的经历来看，在武汉肯定能有坑。

-完-

感谢吴瑞诚的精彩分享。