唯品会亿级数据服务平台实践.docx

唯品会亿级数据服务平台实践数据服务是数据中台体系中的关键组成部分。作为数仓对接上层 应用的统一出入口，数据服务将数仓当作一个统一的DB来访问，提 供统一的API接口控制数据的流入及流出，能够满足用户对不同类 型数据的访问需求。电商平台唯品会的数据服务自2019年开始建设，在公司内经历 了从无到有落地，再到为超过30+业务方提供To B、To C的数据服 务的过程。本文主要介绍唯品会自研数据服务Hera的相关背景、架 构设计和核心功能。背景介绍在统一数仓数据服务之前，数仓提供的访问接入方式往往存在效 率问题低、数据指标难统一等问题，具体而言有以下几个比较突出的 情况：广告人群USP、DMP系统每天需要通过HiveServer以流的方式 从数仓导出数据到本地，每个人群的数据量从几十万到几个亿，人群 数量2w+，每个人群运行时间在30min+，部分大人群的运行直接超 过1h，在资源紧张的情况下，人群延迟情况严重。数仓的数据在被数据产品使用时，需要为每个表新生成一个单独 的接口，应用端需要为每一种访问方式（如Presto、ClickHouse） 区分使用不同的接口，导致数据产品接口暴涨，不方便维护，影响开 发及维护效率。数据在不同的存储时，需要包含clickhouse-client， presto-client 等等第三方 jar 包。不同数据产品中都需要使用一些常用的数据指标，如销售额、订 单数、PV、UV等，而这些数据在不同数据产品的实现口径、实现方 式都不一样，无法形成数据共享，每个数据产品都重复进行相同的指 标建设。因此，在不同数据产品查看相同指标却发现数值不同的情况 下，难以判断哪个数据产品提供的数据是准确的。图1.数据流入流出方式为解决以上问题，数据服务应运而生。目前数据服务的主要优势 有：屏蔽底层的存储引擎、计算引擎，使用同一个APKone service）， 数仓数据分层存储，不同Engine的SQL生成能力，自适应SQL执 行以及统一缓存架构保障业务SLA，支持数据注册并授权给任何调用 方进行使用，提高数据交付效率。通过唯一的ID标识，数据产品可通过ID查阅数据，而非直接 访问对应的数仓表。一方面，指标服务统一了指标的口径，同时也支持快速构建新的数据产品。架构设计数据服务能给业务带来运营和商业价值，核心在于给用户提供自 助分析数据能力。Hera整体架构基于典型的Master/slave模型， 数据流与控制流单独链路，从而保障系统的高可用性。数据服务系统 主要分为三层：应用接入层：业务申请接入时，可以根据业务要求选择数据服务 API(TCPClient)，HTTP以及OSP服务接口(公司内部RPC框架)。数据服务层：主要执行业务提交的任务，并返回结果。主要功能 点包括：路由策略，多引擎支持，引擎资源配置，引擎参数动态组装， SQL Lispengine生成，SQL自适应执行，统一数据查询缓存， FreeMaker SQL动态生成等功能。数据层：业务查询的数据无论在数仓、Clickhouse、MySQL还是 Redis中，都可以很好地得到支持，用户都使用同一套API。智能故据开发应用隈 r&A.V阿心tHJAE瞄gL 项 L J I J L. L J Ifff散据服务r",RHlBM号 IFF，蜻数据层* 冒*il#CIcMgialUs棒人方式 MliKl-KAPIHTTP，冒OSPKK图2.数据服务整体架构图调度系统的整体流程大致包含以下模块：Master：负责管理所有的 Worker、TransferServer、AdhocWorker 节点，同时负责调度分发作业；Worker：负责执行ETL和数据文件导出类型的作业，拉起 AdhocWorker进程（Adhoc任务在AdhocWorker进程中的线程池中 执行），ETL类型的作业通过子进程的方式完成；Client：客户端，用于编程式地提交SQL作业；ConfigCenter：负责向集群推送统一配置信息及其它运行时相关 的配置和SQLParser （根据给定的规则解析、替换、生成改写SQL语 句，以支持不同计算引擎的执行）；TransferServer：文件传输服务。_ff_li5Rnin4VI$K -广 f _ 一 H,/jr，f "图3.数据服务调度流程图主要功能Hera数据服务的主要功能有：多队列调度策略、多引擎查询、 多任务类型、文件导出、资源隔离、引擎参数动态组装、自适应Engine 执行和SQL构建。多队列调度策略数据服务支持按照不同用户、不同任务类型并根据权重划分不同 调度队列，以满足不同任务类型的SLA。多引擎查询数据服务支持目前公司内部所有OLAP和数据库类型，包括 Spark、Presto、Clickhouse、Hive、MySQL、Redis。会根据业务具 体场景和要求，选择当前最佳的查询引擎。多任务类型数据服务支持的任务类型有：ETL、Adhoc、文件导出、数据导入。 加上多引擎功能，实现多种功能组合，如Spark adhoc和Presto adhoc。文件导出主要是支持大量的数据从数据仓库中导出，便于业务分析和处理， 比如供应商发券和信息推送等。具体执行过程如下：用户提交需要导出数据的SQL，通过分布式 engine执行完成后，落地文件到hdfs/alluxio.客户端通过TCP 拉取文件到本地。千万亿级的数据导出耗时最多10min。数据导出在 人群数据导出上性能由原来的30min+，提升到最多不超过3min， 性能提升1030倍。具体流程如下：HCFS集季1 一开恰下莅女件，.突敏童试旃的一1提变眄1 业辑提交文件导地任房近回持果图4.数据服务文件下载流程图资源隔离（Worker资源和计算资源）业务一般分为核心和非核心，在资源分配和调度上也不同。主要 是从执行任务Worker和引擎资源，都可以实现物理级别的隔离，最 大化减少不同业务之间相互影响。引擎参数动态组装线上业务执行需要根据业务情况进行调优，动态限制用户资源使 用，集群整体切换等操作，这个时候就需要对用户作业参数动态修改， 如OLAP引擎执行任务时，经常都要根据任务调优，设置不同参数。针对这类问题，数据服务提供了根据引擎类型自动组装引擎参数，并 且引擎参数支持动态调整，也可以针对特定任务、执行账号、业务类 型来设定OLAP引擎执行参数。自适应Engine执行业务方在查询时，有可能因为引擎资源不足或者查询条件数据类 型不匹配从而导致执行失败。为了提高查询成功率和服务SLA保障， 设计了 Ad Hoc自适应引擎执行，当一个引擎执行报错后，会切换到 另外一个引擎继续执行。具体自适应执行逻辑如下图所示：图5.自适应Engine执行SQL构建数据服务SQL构建基于维度事实建模，支持单表模型、星型模 型和雪花模型。单表模型：一张事实表，一般为DWS或者ADS的汇总事实表。星型模型：1张事实表（如DWD明细事实表）+ N张维表，例 如订单明细表（事实表FK二商品ID） +商品维表（维度表PK=商 品 ID）。雪花模型：1张事实表（如DWD明细事实表）+N张维表+M张 没有直接连接到事实表的维表，例如订单明细表（事实表FK二商品ID） +商品维表（维度表PK二商品ID，FK二品类ID） +品类维 表（维度表PK二品类ID）。图6.SQL维度模型任务调度基于Netty库收发集群消息，系统仅仅使用同一个线程池对象EventLoopGroup来收发消息，而用户的业务逻辑，则交由一个单独 的线程池。选用Netty的另外一个原因是“零拷贝”的能力，在大量数据 返回时，通过文件的形式直接将结果送给调用者。多队列+多用户调度业务需求通常包含时间敏感与不敏感作业，为了提高作业的稳定 性和系统的可配置性，Hera提供了多队列作业调度的功能。用户在提交作业时可以显式地指定一个作业队列名，当这个作业 在提交到集群时，如果相应的队列有空闲，则就会被添加进相应的队 列中，否则返回具体的错误给客户端，如任务队列满、队列名不存在、 队列已经关闭等，客户端可以选择“是否重试提交”。当一个作业被添加进队列之后，Master就会立即尝试调度这个 队列中的作业，基于以下条件选择合适的作业运行：每个队列都有自己的权重，同时会设置占用整个集群的资源总量， 如最多使用多少内存、最多运行的任务数量等。队列中的任务也有自己的权重，同时会记录这个作业入队的时间， 在排序当前队列的作业时，利用入队的时间偏移量和总的超时时间， 计算得到一个最终的评分。除了调度系统本身的调度策略外，还需要考虑外部计算集群的负 载，在从某个队列中拿出一个作业后，再进行一次过滤，或者是先过 滤，再进行作业的评分计算。一个可用的计算作业评分模型如下：队列动态因子二队列大小/队列容量* （1 -作业运行数/ 队列并行度）这个等式表示的意义是：如果某个队列正在等待的作业的占比比 较大，同时并行运行的作业数占比也比较大时，这个队列的作业就拥 有一个更大的因子，也就意味着在队列权重相同时，这个队列中的作 业应该被优先调度。作业权重二1 -（当前时间-入队时间）/超时时间这个等式表示的意义是：在同一个队列中，如果一个作业的剩余 超时时间越少，则意味着此作业将更快达到超时，因此它应该获得更 大的选择机会。Score =作业权重+队列动态因子+队列权重这个等式表示的意义是：对于所有的队列中的所有任务，首先决 定一个作业是否优先被调度的因子是设置的队列权重，例如权重为 10的队列的作业，应该比权重为1的队列中的作业被优先调度，而 不管作业本身的权重（是否会有很大的机率超时）；其次影响作业调 度优先级的因子是队列动态因子，例如有两个相同权重的队列时，如 果一个队列的动态因子为0.5,另外一个队列的动态因子是0.3,那 么应该优先选择动态因子为0.5的队列作业进行调度，而不管作业 本身的权重；最后影响作业调度优先级的因子是作业权重，例如在同 一个队列中，有两个权重分别为0.2和0.5的作业，那么为了避免 更多的作业超时，权重为0.2的作业应该被优先调度。简单描述作业的排序过程就是，首先按队列权重排序所有的队列; 对于有重复的队列，则会计算每个队列的动态因子，并按此因子排序； 对于每一个队列，作业的排序规则按作业的超时比率来排序；最终依 次按序遍历每一个队列，尝试从中选择足够多的作业运行，直到作业 都被运行或是达到集群限制条件。这里说足够多，是指每一个队列都 会有一个最大的并行度和最大资源占比，这两个限制队列的参数组合， 是为了避免因某一个队列的容量和并行度被设置的过大，可能超过了 整个集群，导致其它队列被“饿死”的情况。SQL作业流程用户通过Client提交原始SQL,这里以Presto SQL为例， Client在提交作业时，指定了 SQL路由，则会首先通过访问 SQLParser服务，在发送给Master之前，会首先提交SQL语句到 SQLParser服务器，将SQL解析成后端计算集群可以支持的SQL语 句，如Spark、Presto、ClickHouse等，为了能够减少RPC交互次 数，SQLParser会一次返回所有可能被改写的SQL语句。在接收到SQLParser服务返回的多个可能SQL语句后，就会填 充当前的作业对象，真正开始向Master提交运行。Master在收到用户提交的作业后，会根据一定的调度策略，最 终将任务分发到合适的Worker上，开始执行。Worker会首先采用 SQL作业默认的执行引擎，比如Presto,提交到对应的计算集群运 行，但如果因为某种原因不能得到结果，则会尝试使用其它的计算引 擎进行计算。当然这里也可以同时向多个计算集群提交作业，一旦某 个集群首先返回结果时，就取消所有其它的作业，不过这需要其它计 算集群的入口能够支持取消操作。当SQL作业完成后，将结果返回到Worker端，为了能够更加 高效地将查询结果返回给Client端，Worker会从Master发送的 任务对象中提取Client侧信息，并将结果直接发送给Client，直 到收到确认信息，至此整个任务才算执行完毕。在整个作业的流转过程中，会以任务的概念在调度系统中进行传 播，并经历几个状态的更新，分别标识new、waiting、running、 succeed、failed 阶段。图9. SQL作业处理流程Metrics 采集数据服务搜集两类metrics，一类静态的，用于描述 master/worker/client的基本信息；一类是动态的，描述 master/worker的运行时信息。这里主要说明一下有关集群动态信息 的采集过程及作用。以worker为例，当worker成功注册到master 时，就会开启定时心跳汇报动作，并借道心跳请求，将自己的运行时 信息汇报给master。这里主要是内存使用情况，例如当前worker通 过估算方法，统计目前运行的任务占据了多少内存，以便master能 够在后续的任务分发过程中，能够根据内存信息进行决策master会 统计它所管理的集群整个情况，例如每个任务队列的快照信息、 worker的快照信息、集群的运行时配置信息等，并通过参数控制是 否打印这些信息，以便调试。解决的性能问题数据服务主要解决SLA方面的问题。如人群计算、数据无缝迁 移、数据产品SLA等，这里用人群举例说明如下：人群计算遇到的问题：人群计算任务的数据本地性不好；HDFS存在数据热点问题；HDFS读写本身存在长尾现象。数据服务改造新的架构方案：计算与存储同置，这样数据就不需通过网络反复读取，造成网络 流量浪费。减少HDFS读写长尾对人群计算造成的额外影响，同时减少人群 计算对于HDFS稳定性的影响。广告人群计算介于线上生产任务跟离线任务之间的任务类型。这 里我们希望能保证这类应用的可靠性和稳定性，从而更好地为公司业 务赋能通过数据服务执行人群计算。图10. Alluxio和Spark集群混部基于Alluxio的缓存表同步将Hive表的location从HDFS路径替换为Alluxio路径， 即表示该表的数据存储于Alluxio中。我们使用的方案不是直接写 通过ETL任务写Alluxio表的数据，而是由Alluxio主动去拉取 同样Hive表结构的HDFS中的数据，即我们创建了一个HDFS表的 Alluxio缓存表。由于Alluxio不能感知到分区表的变化，我们开发了一个定时 任务去自动感知源表的分区变化，使得Hive表的数据能够同步到 Alluxio 中。具体步骤如下：定时任务发起轮询，检测源表是否有新增分区。发起一个SYN2ALLUXIO的任务由数据服务执行。任务执行脚本为将Alluxio表添加与HDFS表相同的分区。分区添加完成之后，Alluxio会自动从mount的HDFS路径完 成数据同步。图11. Alluxio缓存表同步人群计算任务上小节介绍了如何让Alluxio和HDFS的Hive表保持数据同 步，接下来需要做的就是让任务计算的Spark任务跑在Spark与 Alluxio混部的集群上，充分利用数据的本地性以及计算资源的隔离 性，提高人群计算效率。人群服务通过调用数据服务执行。数据服务根据底表分区是否同 步到Alluxio决定是否需要下推是用Alluxio表来完成计算。如果 底表数据已经同步到Alluxio，则使用Alluxio表来做为底表计算 人群。依靠数据服务调度系统，通过用户SQL改写以及Alluxio和 Spark计算结点混部模式，人群计算任务提速了 10%30%。小结虽然截至今天，Hera数据服务已经支持了很多生产业务，但目 前仍有很多需要完善的地方：不同engine存在同一个含义函数写法不一致的情况。这种情况 在Presto跟ClickHouse的函数比较时尤为突出，如Presto的 strpos (string, substring)函数，在 Clickhouse 中为 position(haystack, needle, start_pos)，且这些函数的参数顺 序存在不一致的情况，如何更优雅地支持不同engine的差异情况还 需要进一步思考。人群计算采用业界通用的ClickHouse BitMap解决方案落地， 提升人群的计算效率同时扩展数据服务的业务边界。