【AICon】探索RAG 技术在实际应用中遇到的挑战及应对策略!AICon精华内容已上线73%>>> 了解详情
写点什么

唯品会实时计算平台的演进之路

  • 2019-04-19
  • 本文字数:7015 字

    阅读完需:约 23 分钟

唯品会实时计算平台的演进之路

本文由 dbaplus 社群授权转载。

一、实时平台现状


先介绍一下我们整个平台的现状,按计算的话,分为 Storm、Spark、Flink 三个主要的计算引擎,Flink 相应的应用数量目前少一些,不过按照整个计算引擎的发展方式,后续我们还是希望以 Flink 为主做相应的业务推进。

1、核心业务


  • 实时推荐引擎:这块是非常核心的业务,对于大数据来说这些都是个性化推荐、实时推荐;

  • 实时看板:我们称为给到总裁级别、各个业务线的运营和商务的数据,例如各个业务线上的实时销售和用户访问数据;

  • 实时数据,包括 MySQL binlog 数据和用户行为数据:


我们通过 VDP 将 binlog 解析后同步到 Kafka 中,给到后续业务方;


  • 用户行为数据是把 app 或者微信等埋点曝光的数据清洗出来,给后续的实时应用;

  • 金融风控与安全业务的风控;

  • 比价:大家在购买电商商品的时候会比价,我们平台方面也会做,看同样一件商品在其他电商售价多少,以决定我们自己的价格策略;

  • 监控:我们有很多实时的监控系统业务指标,需要利用到实时的指标,比如全站的 PV、UV 等。

2、实时平台的职责

整个实时平台我们可以看两大板块:



  • 实时计算平台:提供基础数据的支持,保证应用的监控数据的稳定性,提供相应的开发支持,包括应用的开发方案评审和引导;

  • 实时基础数据:很多时候我们需要提供基础数据(埋点、Binlog 数据)的清洗打宽,这方面技术性也很强,会汇集到各种业务方。此外,格式的定义、自然的监控也是我们来做,会提供技术平台和技术数据,这些体现着平台的重要职责。

二、实时平台的发展历程

1、早期


整个实时平台的发展过程中,实时这方面也是从 0 到 1,大规模开始进行实时的业务大概是在 2013、2014 年,期间在不断地产生和迭代。


  • Storm 是最早成熟的实时计算框架,最开始也是基于 Storm 做实时应用开发和推广。

  • 早期也在 Storm 尝试去做 SQL 相关的事,那时候想法挺美好,但是因为实时的 SQL 能完成的事情,包括框架的成熟度比较低的情况下,SQL 相互碰撞,步子迈得太大了一点,就死在了沙滩上,我们当时能够真正帮到业务用 SQL 完成的事情还太少。现在相对而言,实时框架上的 SQL(Flink SQL)完善了很多,可以推广使用。

  • 有段时间也做所有实时应用的对接,对我们来说最大的就是资源瓶颈,人员不足,后面还是聚焦平台为主,包括负责部分核心应用。

2、现在

整个实时平台的发展过程经历了几个阶段,具体的技术发展过程如下:



首先是有了 VRC,用于任务管理、监控、告警、数据质量。实时应用都是线上的应用,没有平台统一管控的话,实时应用是否 OK 或数据是否延迟等情况出现,就都不知道。有了实时平台之后,我们就知道实际应用运行到底怎么样。


Storm 比较完善后,计算框架增加 Spark Streaming。


很多用户会用到我们提供基础的数据,比如流量的数据、Binlog 等,并且我们会提供业务上的聚合和打宽。因为很多情况下我们用流量数据时,希望这个数据能包含的维度比较多,最典型的维度是用户过来以后,基本上有对应的设备 ID 转换成唯品会的 ID。


这个设备 ID 关联的用户的性别、姓名,或者用户的基本会员等级、用户偏好等一些基础的数据我们直接关联好,后续做实时任务就简单了很多。


2018 年我们开始推广 Flink 相关的应用,团队内部做的应用全部切到了 Flink。也是从 2018 年 6 月开始,调度开始切换到 Kubernetes 上。我们的实时平台不仅做实时相关的事,也做很多跟机器学习相关的工作,希望机器资源能更好地去共享。


机器资源对于电商来说,会有一个很明显的特征。在 618、双十一这样的大促时,随着数据量的暴增,资源的消耗量也会迅猛增长。


大促结束之后,这些应用根本就不需要这些资源,那么在平常一些时候,这些资源可以给机器学习更多的资源,通过统一资源管理框架调度系统做的这件事情,这是我们切换到 Kubernetes 上的很重要的原因。


我们一直往后面推进,里面也有很多的技术问题需要解决。

3、平台架构

实时平台来说,各家公司的架构基本上差不多。主要的核心数据就是两方面:一是数据源自用户行为的数据;另外就是线上对接的业务数据库的数据。


实时数据到达后,通过计算引擎完成业务逻辑的计算,以及和其他数据的交互,包括支持实时访问的存储系统。数据的结果可以给到实时报表、推荐引擎,也有给到机器学习平台进行模型训练。


整个数据流的过程会经历很多的环节,通过很多次计算的过程,对稳定性和可靠性都有比较高的要求,核心体系架构如下所示:



实时平台提供一个交互式开发的环境,一些简单的逻辑或者可以用 SQL 表达到的;复杂逻辑会使用计算框架的 API 进行开发。


对于任务来说,版本、数据质量以及元数据系统和离线还不太一样。离线的元数据系统都是表,任务很多都是依赖报表来完成。


实时的元数据系统就会考虑到它是在哪;我怎么可以访问到它;它里面数据的格式是怎样的。因为数据可能是 PB 格式,也可能是 JSON,我们会考虑怎么把这个数据格式对业务方透明,在上面做相应的封装来保证业务的方式比较简洁。


实时计算在从早期的荒野到现在逐步完善,我用痛并快乐来形容:痛是因为它开发的难度、开发的要求比离线和其他应用开发要难;快乐在问题解决以后,以及对业务的增长带来成就。

三、实时计算的难点和挑战

1、开发复杂度


首先说一下实时开发的挑战。比如实时所有的数据,在我们的处理中都是以流形式过来的,没有开始和结束的概念,离线的如果用 Spark SQL 去处理数据就有明确的开始结束时间。


就如我们做订单处理,其实这个订单的状态是一直变化的。如添加的购物车、下单、支付、物流和收货以及售后等。


物流之前,订单还会扭转到物流仓库。在包括售后和退货等环节后,这整个订单的持续过程会很久,它随时随地在变化。因为一个月以后订单发生的质量问题,也需要平台来处理。


整个计算链路也比较长,同时涉及到的系统有很多,对计算引擎、存储这些都会要求。


相应地,做实时开发有几个讨论的核心问题:


  • 设置各个组件的并行度是不是合理?是不是可以满足需求?怎么去评估我这是合理的还是说它太大还是太小?

  • 面对计算的结果和内部的计算中间结果数据,用什么来存储?中间是没有状态数据的,这状态数据如果量比较大,并且需要跟外部系统密切交互的时候,如果外部系统响应比较慢或者延迟比较高,就会严重影响实时处理的效率;

  • 异常处理困难,多个流之间的处理很具有挑战性,现在来说,没有一个非常完美的方案能解决多个流之间的关联关系。离线处理相对比较简单,但是在实时处理的时候,问题就复杂了很多;

  • 一致性问题:实时数据要做到 Exactly Once 也是非常有挑战的,对系统有非常高的要求;

  • 数据准确性和校验,很多时候做的实时指标都要跟离线进行对数,如何保证实时和离线一致的,需要考虑到的问题就特别多。

2、技术难度


乱序问题


多个流要关联的话,乱序问题是非常麻烦的。因为很多的数据过来的时候会存在乱序问题。有一条数据可能是早发生的,但是后收到了,因为整个数据上报过来的时候,经历了很多很多的重输,经历了各种设备,很难保证数据是严格有序的。


吞吐和延迟


延迟和吞吐 TPS 通常是矛盾的,你希望 TPS 高,怎么办?


通常加大批处理的模式,我希望一次可以处理一批数据,这样吞吐量就上去了;但是这样延迟就会增加。很多时候面临的具体是什么问题,根据具体问题来给出解决方案。


内部的状态数量


比如计算 UV,UV 的维度会非常多,甚至细到商品粒度。


我们在线的商品几百万的时候,对于商品的 UV,以及跟商品关联的的品牌、档期等,这些交叉之后所需要的计算的指标特别大,可能达到千万级的统计量。


怎么去做状态数据的存储,包括这些数据的可用性,就非常重要了。如果任务处理失败了,怎么快速的恢复等。

四、实时平台的发展方向

对于实时平台来说,这些年的发展都是在做一件事情——降低开发门槛。首先,从平台本身来说,我们做了这几件事情:

1、丰富实时基础数据——实时离线融合


提供核心基础数据


我们做的第一件事是提供核心基础数据。这个很多时候可以降低实时业务后续开发难度。因为前面我们刚刚提到了做多种实时宽表以后,业务做实时应用其实就只是处理单一流。


在前置的任务中,我们把很多特别难搞的问题都解决掉,包括关联到其他维度的数据,多个数据源之间的乱序等,关联好的数据输出到一个流中,后续实时的应用处理单一流就简单了很多。


实时和离线统一


因为大家都知道有一种 Labmda 提法,我实时地算一遍,然后用批处理去补数。


但这会面临两个问题,一是数据需要两个计算框架,工作量大了很多;二是数据重复计算,导致更多的资源消耗。


我们现在来做里面的事情,能实时处理的数据就不会再去离线处理。因为实时处理的实时清洗打宽以后,我们以五分钟的粒度写入 HDFS 的准实时表里面,离线任务就可以启动来使用这些数据了。这样,很多数据的的口径就做到了一致性。


业务方有需求要一个他们认为的实时数据指标,首先作为平台来说,就会问你一个直白的问题:你希望这个实时是什么概念?是秒级还是分钟级?


因为对于实时的时间密度大家理解是不一致的,如果在分钟级以上我们会根据这些数据通过准实时或者离线的方案来完成数据的输出。因为离线 SQL 的开发成本是最低的,可能我直接写 SQL 就能完成。


相应的,如果你真的要求数据时间非常短,那要考虑真正实时的应用怎么开发。

2、统一计算资源:基于 Kubernetes 调度实时和 AI 平台资源


前面我们也提到了在管理计算资源的时候,如果做到 Storm、Flink 或者 Spark 等任务通过 Kubernetes 统一资源的调度管理后,能快速地对应用扩收容。实时平台的管理实时应用接近 800 个,所以需要快速地提升自身应用的能力和变更。

3、新的开发模式


实时平台现在推广基于 Streaming SQL 的开发模式,底层的计算引擎以 Flink 和 Spark Structured Streaming 为主。


因为不管是 Flink 还是 Spark Structured Streaming,都支持状态存储,便于错误恢复和可靠性保障。


第二个是相应的,我们整个 API 要更高级一些。相应的,写一些代码的时候,也会有非常大的简化。


整个来说,开发平台也在预研实时的 notebook 开发环境,你可以完成相应的实时应用的开发,整个模式都是在替换 Storm 的节奏。

4、效率提升

在我们的实践中,举一些使用 Flink 带来的显著好处的案例。


UV 计算



在以 Storm 为计算框架的时候,状态存储通常是在 Redis 中,这样会需要大量的 Redis 资源。


现在完全可以用 Flink 来做,它的好处是我们有很多的计算任务,通过这样的转化,它计算的稳定性、可靠性得以提升,计算资源消耗降低 2/3。


以我们自己 UV 中品牌日这个场景为例,同时会有上百个品牌日在线,因此每个品牌日的指标需要全部的预算指标。在 Storm 为计算的时候甚至用到了 240 个以上的 Worker 节点,并且配合了 2T 的 Redis 存储。


切到了 Flink 以后,因为 Flink 所有的状态都是在内存中,这个过程中少了跟 Redis 的交互,效率会有非常大的提升。同时整个的依赖组件越少,系统稳定性就会越高。


埋点数据落地



我们很多实时的在 Kafka 里面的数据通过 Spark Streaming 写入到 HDFS 中,然后通过 Hive、Spark SQL 来访问。


在使用 Flink 里带的 Bucket Sink 模式,只需要十分之一的资源就达到了原来 Spark Streaming 就能达到相应的吞吐量,并且延迟也大幅度的降低。

5、统一数据源

1)UDM 架构


当前实时系统,机器学习平台要处理的数据分布在各种数据存储组件中,如 Kafka、Redis、Tair、HDFS 等等,如何方便高效的访问、处理、共享这些数据是一个很大的挑战,对于当前的数据访问和解析常常需要耗费很多的精力。主要的痛点包括:


  • 对于 Kafka,Redis,Tair 中的 binary(PB/Avro 等格式)数据,使用者无法快速直接的了解数据的 schema 与数据内容,采集数据内容及与写入者的沟通成本很高。

  • 由于缺少独立的统一数据系统服务,对 Kafka,Redis,Tair 等中的 binary 数据访问需要依赖写入者提供的信息,如 proto 生成类,数据格式 wiki 定义等,维护成本高,容易出错。

  • 缺乏 relational schema 使得使用者无法直接基于更高效易用的 SQL 或 LINQ 层 API 开发业务。

  • 无法通过一个独立的服务方便的发布和共享数据。

  • 实时数据无法直接提供给 Batch SQL 引擎使用。

  • 此外,对于当前大部分的数据源的访问也缺少审计,权限管理,访问监控,跟踪等特性。


UDM(统一数据管理系统)包括 Location Manager、 Schema Metastore 以及 Client Proxy 等模块,主要的功能包括:


  • 提供从名字到地址的映射服务,使用者通过抽象名字而不是具体地址访问数据。

  • 用户可以方便的通过 Web GUI 界面方便的查看数据 Schema,探查数据内容。

  • 提供支持审计,监控,溯源等附加功能的 Client API Proxy。

  • 在 Spark/Flink/Storm 等框架中,以最适合使用的形式提供这些数据源的封装。


UDM 的体系架构如下图所示:



2)基于 UDM 的开发模式


在基于 UDM 的基础上,实时和离线可以统一开发模式,其示例代码如下:



从示例可以看出,不管 Flink 还是 Spark 都是统一的开发模式,而且代码非常简洁。

Q & A

Q1:我们也是做这个的,你的大数据平台上所有的组件都是容器化的还是混合的?是属于计算类的?


A1:现在大数据的所有组件要容器化很难,存储我们还没有做容器化相关的事,更多的是做偏计算引擎,就是我计算容器化掉,能给到很多更明确的意义。


比如说我很难告诉你一个在 YARN 上的任务到底吃掉了多少 CPU,到底系统有没有问题。那现在容器化后就可以非常好地监控应用的资源消耗,也非常容易判断资源是否使用合理。


第二个,容器化以后,可以开放一些更高的权限给应用的开发者,我可以允许你进到容器里去做一些以前需要高权限才能做的事,比如说要去做一些性能分析问题的时候,在物理机上不太可能给你比较高的权限的,你进去看到的包括其他开发者的应用,操作不当可能把组件都搞挂了。


但是在容器环境下,你登录到容器里面,搞死了就是你自己把自己的应用搞死了,别人的你根本就看不到。


我们目前是计算类的。对于大数据无外乎就计算和存储两大类。存储容器化目前来说,优势不明显,反而会带来更多问题。


现在对于大数据来说,一个趋势是以前我们希望计算和存储在一起,现在由于网络带宽越来越高,我们偏好计算和存储分离。因为大的互联网公司基本上都是从至少单机万兆以上的网络。相对而言,数据的本地性重要性就没有那么高。存储和计算分离以后,计算节点扩展就更加容易。


Q2:我想问下,如果在流计算的统计过程中要维护很多状态,比如有几十亿的要维护,根据您刚刚提到的技术难点,能讲讲是怎样解决的吗?下一个问题是各个节点写入不均衡,怎么去监控,怎么解决?怎么样看数据倾斜呢?


A2:第一个问题,如果以 Storm 来说,可以做两级存储,依赖 Storm 我肯定会在自己 Worker 的 JVM 内部做一级缓存。


达到一定的时间和一定量之后,刷到外部的二级存储里面。计算维度数据的统计指标非常庞大的时候,如果直接往外部输,说得直白一点,再强悍的外部存储的性能都扛不住。你也提到了上亿的时候,每秒输出的数据都是几百万 QPS,基本上没有一个存储的引擎可以在比较合理的规模来支持这个量级(除非依赖更大规模的分布式存储引擎)。


因此一般都是两级存储,到了一定时间或者一定状态数量之后把它刷新到外部的存储引擎。这样就有一个问题,你会丢掉最近未刷新出去的那批数据,这就要求你可以从上一个刷新出去的数据状态支持恢复,或者重复处理。


Storm 完全依赖自己的状态管理,数据出现失败,或者恢复的时候,需要自己处理比较多的事情。Flink 相对而言就好很多,支持从 checkpoint 或者 savepoint 直接恢复。


第二个问题对于数据倾斜来说,基本上所有的数据倾斜的处理都是两阶段处理。即先做一级计算汇总,一级计算之后再做二级处理;或者明确知道哪些数据分布特别不均匀,把这些数据单独处理。


关于怎么样看数据倾斜,观察每一级 Task 往下一级 Task 写出的数据量的分布就可以了。在我看来,数据量最大的比最小的高了十倍就肯定是倾斜,当然也可以计算标准差。


Q3:您之前说唯品会是电商方面的,我们公司也是电商方面的。想问关于订单方面的状态您是怎么维持它的瞬时性的?


A3:对于订单,或者还是其他的业务数据,就是两种场景,一种场景要求我们自己做的工具,我们对应的同一张表,哪怕是分表的数据,同一个表或者分库分表写到同一个 Kafka 的 Topic 里面,并且根据分表做 Partition 映射,这样可以保证局部一致性,即同一个物理表的数据是严格有序的。要保证全局一致性,其实是保证不了的。


但是,对业务来说,局部的一致性就 OK 了,同一个订单只能写到同一个分表上。这样在读取数据的时候,就能做到按相应顺序消费。


第二种,像一些特殊的场景,可能还是会依赖我们自己把秒级数据抽取,即高频的以 SQL 方式去拉取最新的数据写入到 Kafka 中。


很多时候我们做订单统计,也会缓存所有的订单计算过程一些中间状态。最典型的,订单的中间状态的变化是很复杂的一件事。我下单时第一件事是拆单,一个订单能买三件商品,我们有复杂的拆单逻辑,即将这个母单会拆成三个子单,或者两个子单。


因为对应了不同的物流仓库,也可能有下的海淘的商品。在签收之前的过程中你可能又取消了一个子单,或者把三个子单都取消,那这个订单就彻底取消掉了。


因此我们算单数的时候,就要全部减掉。整订单的各种状态我们会把它缓存起来。因为像我们的订单大概有接近 40 个状态,只是有些状态可能不见得会在统计的时候用到。

作者介绍

王新春,唯品会高级经理、数据平台实时团队高级架构师,主要负责实时计算平台、实时数据、实时报表和机器学习平台等业务;曾任美团点评(原大众点评)数据平台高级架构师,负责从零开始搭建实时计算平台及数据平台工具体系开发和建设等工作。


2019-04-19 08:0013330
用户头像
dbaplus社群 数据连接未来

发布了 176 篇内容, 共 70.6 次阅读, 收获喜欢 599 次。

关注

评论

发布
暂无评论
发现更多内容

Redis系列(七):缓存只是读写回种这么简单吗?如果是,那么请你一定看看这篇文章!

z小赵

redis 分布式 高并发系统设计

LeetCode 1052. Grumpy Bookstore Owner

liu_liu

算法 LeetCode

Unix路径是如何简化算法,架构师性能优化 John 易筋 ARTS 打卡 Week 10

John(易筋)

ARTS 打卡计划

Java架构-Java代码规范那些事

我是苞谷

Java

ARTS WEEK6

紫枫

ARTS 打卡计划

【API进阶之路】无法想象!大龄码农的硬盘里有这么多宝藏

华为云开发者联盟

容器 层次 API 网关 华为云

应用程序研发之基础知识分层与进化

superman

设计模式之外观模式解析

Seven七哥

程序员 设计模式 外观模式

华为云FusionInsight MRS融合大数据平台进阶之路

数据湖洞见

大数据 新特性 FusionInsight 华为云 智能数据湖

区块链如何切入供应链金融市场?

CECBC

安全系列之——RSA的前世今生

诸葛小猿

安全 加密解密 非对称加密 rsa

实战案例丨ModelArts在数据标注、数据过滤上的应用技巧:自动分组

华为云开发者联盟

人工智能 数据 图像识别 图片 分类

Golang新手常犯错误之【循环迭代篇】

卓丁

常见错误 引用迭代 Go 语言

浅析区块链如何改变生活

CECBC

数字银行 供应链 身体监测 资产管理

股权交易中心+区块链试点将开始

CECBC

防篡改 股权交易 可追溯 信息存证

一个好用的工作生活平衡方式

霍太稳@极客邦科技

week8 作业

Shawn

职场求生攻略答疑篇之 1 —— 加班沉思录

臧萌

程序员 加班

高并发系统设计要点

南方有乔木兮

Java

正则表达式基础详解

懒猫

Java 正则表达式 大前端 正则

面试题:Java 中的 ==, equals 与 hashCode 的区别与联系

简爱W

性能测试 + 操作系统 + 锁

鲁米

[POJ 1002] 487-3279 C++解题报告

一直AC一直爽

POJ ACM

架构师训练营第八周作业

张明森

7个获取访问者真实IP的方法,速学!!!

华为云开发者联盟

nginx 大前端 网站 IP 服务器

一文了解JDK12 13 14 GC调优秘籍-附PDF下载

程序那些事

GC JDK14 秘籍 JDK12 JDK13

Java架构-不要成为项目风险的奴隶

我是苞谷

Java

Kafka两个高性价比的参数调优

我是个bug

Java 大数据 kafka

推荐一款技术人必备的接口测试神器:Apifox

狂师

测试 测试驱动开发实战营 接口测试 测试框架

如何去学好JS的8条小建议

华为云开发者联盟

html 编程 大前端 js 代码

Go: 通过代码学习 Map 的设计 — Part II

陈思敏捷

map Go 语言

唯品会实时计算平台的演进之路_5G/IoT_dbaplus社群_InfoQ精选文章