Apache Flink在唯品会的实践

唯品会实时平台现状

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

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

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

在架构设计上，包括两大数据源。一种是在App、微信、H5等应用上的埋点数据，原始数据收集后发送到在kafka中;另一种是线上实时数据的MySQL Binlog日志。数据在计算框架里面做清洗关联，把原始的数据通过实时ETL为下游的业务应用(包括离线宽表等)提供更易于使用的数据。

Flink在唯品会的实践

场景一：Dataeye实时看板

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

以UV计算为例，首先对Kafka内的埋点数据进行清洗，然后与Redis数据进行关联，关联好的数据写入Kafka中;后续Flink计算任务消费Kafka的关联数据。通常任务的计算结果的量也很大(由于计算维度和指标特别多，可以达到上千万)，数据输出通过也是通过Kafka作为缓冲，最终使用同步任务同步到HBase中，作为实时数据展示。同步任务会对写入HBase的数据限流和同类型的指标合并，保护HBase。与此同时还有另一路计算方案作为容灾。

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

在将计算任务从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。字典表的变化是由离线批处理任务引起的，目前的做法是使用ContinuousFileMonitoringFunction和ContinuousFileReaderOperator定时监听HDFS数据变化，不断地将新数据刷入，使用最新的数据去做join实时数据。

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

Flink On K8S

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

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

唯品会在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脚本里面需要设置的环境变量设置说明：

（编辑：核心网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!