资源消耗降低2/3,Flink在唯品会实时平台的应用

  1. 云栖社区>
  2. dbaplus社群>
  3. 博客>
  4. 正文

资源消耗降低2/3,Flink在唯品会实时平台的应用

技术小能手 2018-11-02 14:24:36 浏览1089
展开阅读全文

本文主要内容包括以下几个方面:

 ●  唯品会实时平台现状;
 ●  Flink在唯品会的实践;
 ●  Flink On K8S;

 ●  最新项目进展。

一、唯品会实时平台现状

目前在唯品会,实时平台并不是一个统一的计算框架,而是包括Storm、Spark、Flink在内的三个主要计算框架。由于历史原因,当前在Storm平台上的job数量是最多的,但是从去年开始,业务重心逐渐切换到Flink上面,所以今年在Flink上面的应用数量有了大幅增加。

实时平台的核心业务包含八大部分:

 ●  RTRS(实时推荐引擎):实时推荐作为电商的重点业务,包含多个实时特征;
 ●  Dataeye(大促看板):包含各种维度的统计指标(例如:各种维度的订单、UV、转化率、漏斗等),供领导层、运营、产品决策使用;
 ●  实时数据清洗:从用户埋点收集来数据,进行实时清洗和关联,为下游的各个业务提供更好的数据;
 ●  互联网金融;
 ●  安全风控;
 ●  与友商比价;
 ●  Logview、Mercury、Titan作为内部服务的监控系统;
 ●  VDRC实时数据同步系统等。

共计有1400台机器和600+应用。

97bcf823e35f7cb1ec91cd5ba9c579fb95188234

图1 实时核心业务

实时平台的职责主要包括实时计算平台和实时基础数据:

 ●  实时计算平台在Storm、Spark、Flink等计算框架的基础上,为监控、稳定性提供了保障,为业务开发提供了数据的输入与输出;
 ●  实时基础数据包含对上游埋点的定义和规范化,对用户行为数据、MySQL的Binlog日志等数据进行清洗、打宽等处理,为下游提供质量保证的数据。

在架构设计上,包括两大数据源:

 ●  一种是在App、微信、H5等应用上的埋点数据,原始数据收集后发送到Kafka中;

 ●  另一种是线上实时数据的MySQL Binlog日志。数据在计算框架里面做清洗关联,把原始的数据通过实时ETL为下游的业务应用(包括离线宽表等)提供更易于使用的数据。

24f39351b291ca9989d88c1eb796c5f2163a520b

图2 实时平台整体架构

二、Flink在唯品会的实践

场景一:Dataeye实时看板

Dataeye实时看板,支持需要对所有的埋点数据、订单数据等进行实时计算。具有数据量大的特点。并且需要统计的维度有很多,例如全站、二级平台、部类、档期、人群、活动、时间维度等,提高了计算的复杂程度,统计的数据输出指标每秒钟可以达到几十万。

以UV计算为例,首先对Kafka内的埋点数据进行清洗,然后与Redis数据进行关联,关联好的数据写入Kafka中;后续Flink计算任务消费Kafka的关联数据。

通常任务计算结果的量也很大(由于计算维度和指标特别多,可以达到上千万),数据输出也是通过Kafka作为缓冲,最终使用同步任务同步到HBase中,作为实时数据展示。

同步任务会对写入HBase的数据限流和同类型的指标合并,保护HBase。与此同时,还有另一路计算方案作为容灾。

880a1e5148a80eadf556f2566349fa131a9f33c7

图3 Dataeye计算任务架构

在以Storm为计算引擎中的进行指标计算时,需要使用Redis作为中间状态的存储。而切换到Flink后,Flink自身具备状态存储,节省了存储空间。由于不需要访问Redis,也提升了性能,整体资源消耗降低到了原来的1/3。

5790140297581cfd3e49a9520a2a6337e9e884e2

图4 Storm VS Flink计算架构

在将计算任务从Storm逐步迁移到Flink的过程中,对两路方案先后进行迁移,同时将计算任务和同步任务分离,缓解了数据写入HBase的压力。

切换到Flink后也需要对一些问题进行追踪和改进。对于FlinkKafkaConsumer,由于业务原因,对Kafka中的Aotu Commit进行修改及对Offset的设定需要自己实现支持Kafka集群切换的功能。对不带Window的state数据需要手动清理。还有计算框架的通病——数据倾斜问题需要处理。同时对于同步任务追数问题,Storm可以从Redis中取值,Flink只能等待。

场景二:Kafka数据落地HDFS

之前都是通过Spark Streaming的方式去实现,现在正在逐步切换到Flink上面,通过OrcBucketingTableSink将埋点数据落地到HDFS上的Hive表中。

在Flink处理中,单Task Write可达到3.5K/s左右,使用Flink后资源消耗降低了90%,同时将延迟30s降低到了3s以内。目前还在做Flink对Spark Bucket Table的支持。

场景三:实时的ETL

对于ETL处理工作而言,存在的一个痛点就是字典表存储在HDFS中,并且是不断变化的,而实时的数据流需要与字典表进行join。

字典表的变化是由离线批处理任务引起的,目前的做法是使用ContinuousFile MonitoringFunction和ContinuousFileRea derOperator定时监听HDFS数据变化,不断地将新数据刷入,使用最新的数据去做join实时数据。

我们计划用更加通用的方式去支持Hive表和Stream的join,实现Hive表数据变化后数据自动推送的效果。

三、Flink On K8S

在唯品会内部有一些不同的计算框架:有实时计算的,有机器学习的,还有离线计算的,所以需要一个统一的底层框架来进行管理,因此将Flink迁移到了K8S上。

在K8S上使用了思科的网络组件,每个Docker容器都有独立的IP,对外也是可见的。实时平台的融合器整体架构如下图所示:

c22619e899feaf35acaedc6e8affaf7c91c3a4af

图5 Flink On K8S架构

唯品会在K8S上的实现方案与Flink社区提供的方案差异还是很大的。唯品会使用K8S StatefulSet模式部署,内部实现了cluster相关的一些接口。一个job对应一个mini cluster,并且支持HA。

对于Flink来说,使用StatefulSet的最大的原因是pod的hostname是有序的;这样潜在的好处有:

 ●  hostname为-0和-1的pod可以直接指定为jobmanager;可以使用一个statefulset启动一个cluster,而deployment必须2个;Jobmanager和TaskManager分别独立的deployment;
 ●  pod由于各种原因fail后,由于StatefulSet重新拉起的pod的hostname不变,集群recover的速度理论上可以比deployment更快(deployment每次主机名随机)。

镜像的docker entrypoint脚本里面需要设置的环境变量设置说明:

d9c8d8f733857ce49161af4bcd6beb6512ee8d94

对应Flink集群所依赖的HDFS等其他配置,则通过创建configmap来管理和维护。

kubectl create configmap hdfs-conf --from-file=hdfs-site.xml --from-file=core-site.xml
四、最新项目进展

当前实时系统、机器学习平台要处理的数据分布在各种数据存储组件中,如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的整体架构如下图所示:

9c5b11fab14a3d1a3a89d9658abc204911184993

图6 UDM架构

UDM的使用者包括实时、机器学习以及离线平台中数据的生产者和使用者。在使用SQL API或Table API的时候,首先完成Schema的注册,之后使用SQL进行开发,降低了开发代码量。

以Spark访问Kafka PB数据的时序图来说明UDM的内部流程:

d31c7a0e6dcadf1aedcef24341c5caf03644ed32

图7 UDM使用时序图

在Flink中,使用UDMExternalCatalog来打通Flink计算框架和UDM之间的桥梁,通过实现ExternalCatalog的各个接口及实现各自数据源的TableSourceFactory,完成Schema和接入管控等各项功能。

Q & A

Q:能详细描述一下“实现Hive表数据变化之后,数据自动推送的效果”是指什么吗?

A:就是Hive表的数据更新后,可以在Flink里面缓存的数据也可以更新掉。这样jion的时候,就可以关联最新的数据了。目前来说,还是需要一些自己开发工作的,官方框架不支持。


原文发布时间为:2018-11-2

本文作者:王新春

本文来自云栖社区合作伙伴“DBAplus社群 ”,了解相关信息可以关注“DBAplus社群”。

网友评论

登录后评论
0/500
评论
技术小能手
+ 关注
所属云栖号: dbaplus社群