VIPKID实时计算技术实践

2020 年 10 月 15 日 DataFunTalk

文章作者：甄国有@VIPKID

内容来源：Flink中文社区

导读： 对于以互联网形式的在线教育来说，实时计算应用的场景非常之多，如何通过数据计算来更快、更高效地反馈学习效果保证课程质量是在线教育领域不断探索的主题。本文将从以下四个部分分享，与大家探讨在直播上课过程中如何通过实时计算来提高人效以及系统处理能力。

VIPKID 介绍
核心业务场景
技术实现
总结

VIPKID 介绍

VIPKID 是一家在线青少儿英语教育平台，成立七年以来，公司坚持以赋能教育，启迪未来为使命，专注于一对一的线上教学模式，采用 100% 的纯北美外教，学员遍布 63 个国家和地区。

截止目前，仅付费学生规模超 70 万人，单日一对一课量超 10 万节，高峰时段课程并发最高达到 3.5 万节。拥有覆盖了全球 35 个国家的 5 条跨海专线，在 16 个国家、55 个城市完成数据中心传输节点布局，能够根据实时动态在一分钟内完成智能切换[1]。

核心业务场景

主要场景介绍

在一对一（一个老师和一个学生）模式的上课过程中，老师通过直播的形式以课件为辅助进行授课，互动的形式不仅包括直观的声音和视频还有聊天室以及在课件上写字划线拖动动作等，整个课程中涉及多个组件模块。

各模块以协同依赖的方式提供服务，其中任意环节发生的事件对老师和学生都要做到可见和同步，如老师可看见学生在教室才能开始上课、学生可听见老师说话、学生可看到老师翻页课件等才能继续正常上课直到结束。

在大规模网络教学中，流媒体实时互动直播和消息实时数据传输严重依赖用户设备和网络，数据体量大，尤其我们是跨海传输的情况下变得非常棘手，对于网络稳定性有着非常苛刻的要求。

与大班网课直播相比，1v1 更注重互动，所以对问题的容忍度极低，任何一方的问题都会影响上课体验。其中场景之一为当出现网络等异常问题时，用户就会点击”Help“按钮进行求助，此时需要监课人员（以下简称“FM”，来自 Fireman 缩写）立刻介入处理，这对服务人员的规模和操作实时性有较大的需求。

当前业务痛点

目前在只有人工处理用户 Help 的模式下，由于日均 Help 请求量大（约占总课程的10%），人均监课量大，同时从接收到请求到监课人员介入处理问题也需要辗转多个流程，会有以下问题：

问题处理不及时，用户容易等待，阻断上课，带给用户体验差；
人工处理效率低，课量增加以及大规模突发情况下，导致 FM 团队规模增加，需要更多人力;
有些用户出了问题，没有联系监课人员的话，问题被隐藏；

技术实现

为了解决上文提到的业务痛点问题，经过各环节业务特征提取及梳理，我们设计了一种通过实时计算来产出业务标签，并应用标签数据进行自动监课来解决用户 Help 的方案。下文将重点描述整个方案的技术实现细节：涉及到数据体系建设、自动化业务系统建设、核心问题与优化以及最终收益效果：

数据体系建设：介绍用于支撑整个实时计算的 Vlink 数据平台、当前场景下相关业务数据采集和业务标签数据计算，是业务实现的支撑
自动化业务系统：介绍如何应用实时数据流来解决当前业务痛点
问题与优化：介绍实现过程中碰到的业务和技术问题以及解决方案
收益效果：介绍最终获得的收益成果

数据体系建设

整个数据体系建设的初衷是解决数据从哪里来、数据的业务逻辑是什么、如何计算、如何统一管理以及赋能更多场景，解决更多业务问题。

Vlink 数据平台：介绍一站式数据平台，提供数据接入明细：
a.数据来源；
b.数据的业务含义；
c.数据打点规律，提高开发接入效率，解决上下游不明确问题；
业务数据采集：介绍当前场景下的业务数据采集；
业务数据计算：介绍如何应用Flink来计算复杂逻辑的业务数据；

■ Vlink 数据平台

Vlink 数据平台是基于在 Flink Streaming Job 开发过程中一些问题的反思后，借鉴服务端开发上线流程，以研发人员为中心的提高开发效率，降低维护成本为出发点而设计研发的系统，并支持数据采集管理、打点接入管理、打点测试集成等功能。

主要功能点

1. 交互式运行作业

除 Flink Sql 外，业内对于 Streaming 类型的作业提交运行方式还是和官方提供的上传 Jar 包一样，打包 -> 等待并关注 -> 上传 -> 等待并关注 -> 运行。我们联合运维团队，提供一键打包部署功能，可设置 AutoRun 在部署成功后自动运行。

2. 批量执行操作：

在部分场景下需要部分或全量作业重启，当作业量很大时，是个费时费力的过程，而且比较容易出错，那批量构建、停止与运行就变得很轻松，如：

某一类作业逻辑更新；
三方依赖库的升级更新；
集群升级；

3. SP 功能：交互式创建运行 SavePoint。

4. 血缘关系图：体现数据从打点到最终产出这一链路的上下游关系。

从图中可以清楚地知道处理程序 P1、P2 和 P3 的输入输出。

5.其他功能:

版本控制；
支持交互式开发 Flink SQL Job（仅支持 Kafka）。
数据 Schema 查询

开发约束

Flink 作业开发过程中，我们发现核心逻辑是在 pipeline 过程中的 Function，同时有大量重复的逻辑 Function，比如，作业上下文配置、添加 Source 以及设置 WaterMark 等，所以我们抽出了各层的逻辑封装成组件，并做了一些开发约束，让开发者只关注核心逻辑。

1. 提供 'AbstractJobModel'，统一 Schema 化输入数据：

private[garlic] trait AbstractJobModel extends Serializable {  def tm: Long // event time 事件时间  def ingestion: Long // ingestion time 摄入到到Flink系统时间  def f: Boolean // for filter data that is useless 不符合条件要被过滤的数据  def unNatural: Boolean // filter future data “超自然”数据}

unNatural：因各端系统时间不统一而造成的时间戳大于当下时间的数据，我们称其为“超自然”数据，在时间处理语义 EventTime 时需要特别关注。

2. 提供统一的灵活的 Kafka Source 初始化方式

/** *  指定消费时间戳初始化方法 */def initSourceWithTm[T](deserializer: AbstractDeserializationSchema[T], topics: Array[String], tm: Long): SourceFunction[T]
/** *  指定消费时间戳和Kafka Server初始化方法 */def initSourceWithServerAndTm[T](deserializer: AbstractDeserializationSchema[T], topics: Array[String], servers: String, tm: Long): SourceFunction[T]
/** *  通用初始化方法 */def initSource[T](implicit deserializer: AbstractDeserializationSchema[T], topics: Array[String], servers: String, tm: Long = 0L): SourceFunction[T]

3. 多形式 Sink Function

sinkFilteredDataToKafka：不符合规则或异常被过滤。
sinkUnnaturalDataToKafka：超自然数据。
sinkLateDataToKafka：乱序数据应延迟而被 Window Function 丢弃。
sinkDataInAndProcessToKafka：每条数据的摄入时间和处理时间。

4. 支持常用的三方连接组件

Kafka
Hbase
ES
JDBC

■ 业务数据采集

数据采集是整个数据处理架构的基础且重要环节，数据采集的实时性和准确性将直接影响到上层业务，采集方式有间接上传文件的方式和直接 Http 打点的方式。

事件数据埋点涉及到移动端、PC 端和服务端，以进教室为关键事件点：

用户发起进教室流程：加载 SDK 后，请求服务和网关，然后初始化服务组件流媒体、消息通道和动态课件，当所有组件都没有异常时才表示进教室成功了，否则继续重试逻辑直到进教室失败或成功；
进教室成功后，课程在正常进行中时，服务组件持续提供服务并实时上报数据。

整体而言埋点上有问题标签也有正常标签，按照进教室事件和组件类型，由粗到细可分一级、二级和三级。

进教室标签，用户有0到多次进教室记录，因某一组件初始化失败而进不了教室和进教室过长，以及进教室成功。
流媒体标签，主要有音视频卡顿、听不见彼此和看不见彼此以及音视频正常数据，数据打点百毫秒级别。
动态课件标签，主要有课件加载失败、课件动作不同步和无法划线拖动。

■ 业务数据计算

本次业务计算对实时性要求极高，在技术选型上以 Flink 为主[2]，天级别的离线数据分析会以 Spark 为主。

标签计算是整个自动化处理的关键点，指标计算的速度代表着系统能处理的速度，数据来自多个业务流，结合当前的业务场景下，比较典型的计算场景有：

a. 基于事件时间的多流 union

 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val stream = env.addSource(singleSource).name("signal")       .union(env.addSource(avSource).name("av"))      .union(env.addSource(dbySource).name("dby"))      .union(env.addSource(enterSource).name("enter"))
      .filter(_.f)      .filter(_.unNature)      .assignTimestampsAndWatermarks(new DummyEventTimePunctuWaterMarks[InlineInputEventForm](6 * 1000))      .filter(m => *** ).name("***")
    val ***Streaam = stream      .filter(f => *** )      .keyBy(key => *** )      .window(TumblingEventTimeWindows.of(Time.milliseconds(30 * 1000L)))      .sideOutputLateData(***lateOutputTag)      .apply( ***WindowFunction)
    sink***ToKafka(***Streaam, ***name, recordFilter60s, ***kafkaSink, recordTmKafkaSink)

注：*** 表示业务脱敏处理（下同）

b. 多流 Join

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    ***省略部分逻辑代码***
    val ppt***JoinStream = ***Stream      .coGroup(***Stream)      .where(lb => ***)      .equalTo(lb => ***)      .window(SlidingEventTimeWindows.of(Time.milliseconds(30000), Time.milliseconds(15000)))      //.sideOutputLateData(***LateOutputTag)      .apply(ppt***CoGroupWindowFunction)
    sink***StreamToKafka(ppt***JoinStream, ***name, recordFilter60s, ***kafkaSink, recordTmKafkaSink)

coGroup 算子在目前的版本（1.7.2 及以上）不支持迟到数据输出，已经向社区提了相关 Jira[3]。

c. 异步加载维度数据

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  AsyncDataStream.unorderedWait(      stream,      syncGet***Function(),      500L,      TimeUnit.MICROSECONDS    )

另外，维度数据计算时，根据实效性通过 GuavaCacheBuilder[4] 进行热数据缓存。

自动化业务系统

我们通过梳理上课中各关键环节点遇到的问题，从业务上，提出在用户发起 Help 后，且 FM 介入前做一层实时自动化服务的业务解决方案。

从技术上，该自动化业务系统构建在整个数据体系之上，以上课过程中实时标签数据为基础，然后由标签系统应用标签数据流通过预检、自检等手段自动化或半自动化来处理问题。对于系统不能处理的问题，则转人工处理。

首先，课程中的问题有两种上报方式：

被动等用户发起 Help
主动探测问题标签流

然后验证逻辑模块过滤掉无效问题，如无效 Help、重复 Help、请求过期、FM 已经介入以及特殊问题等，还有对于标签体系无法覆盖的问题（比如噪音），则直接转 FM 人工处理。

若请求通过了验证模块且系统可自动处理，则自检处理系统可尝试进行切线，然后进行切线验证并将切线标记放入待处理队列 Pending，Pending 验证阶段实时获取正常标签流以内反馈的方式检测是否恢复正常。

问题与优化

整个业务场景对实时性要求极高，同时还要保证准确性以及需要知道每一条数据的来龙去脉。对于没有正确计算到的 Case 要给出具体计算明细，比如数据在哪一层耗时导致到达处理引擎时间过长、什么环节处理耗时过多、因为乱序哪些数据被丢掉了、如何提高加载维度信息速度、如何巧妙提高系统处理量同时使用的计算资源又少、如何处理“超自然”数据等（详细见“Vlink 数据平台”）。

数据质量良莠不齐，指标不一致：整个数据埋点涉及 3 个部门跨 11 个团队，没有统一口径。通过 Vlink 数据平台按业务层级统一管理数据指标、端版本控制和验证流程；
实时计算下获取维度信息造成对 DB 库压力：a、在业务允许的前提下，通过小窗口聚合数据，减少查询次数；b、根据数据时效性增加缓存；
无课程数据时“造数据”导致数据量翻倍：在串行逻辑下，前置多窗口且窗口大小与核心逻辑窗口大小保持一致，指定与 TaskManager*2 的分片数，预处理获取课程维度信息“造数据”再 shuffle 给下游核心窗口逻辑处理。