揭秘海尔实时计算平台：多业务场景下的技术选型与实践
更多精彩文章请添加微信“AI 前线”（ID：ai-front）
今天我给大家带来的内容是海尔实时计算平台的技术选型与实践。
海尔电器是海尔企业的全资子公司，它包含的业务板块非常多，包括物流、快递柜、健康水站、跨境电商等，其中主要是物流板块。海尔的物流板块在业界属于大型物流，并且配装一体，即配送和安装一体化。实际上，在物流行业里，它的数据量是非常大的，各个环节都需要收集大量数据，并进行数据分析。这么多的业务数据里不仅有大量的历史内容，也包括每天都在产生的新业务数据。海尔电器在很早的时候就有关于大数据方面的规划，是一个逐步建设的过程，从最初依赖 Oracle 数据仓库来建设，逐步过渡到以 Hadoop 为核心，再逐步发展到实时计算。
正如大家所知，实时计算相比离线计算，所涉及的要求、复杂性以及相关技术都更多，我今天会将相关技术与实践经验相结合与大家分享。我的介绍主要分成三部分，第一部分是实时计算平台的背景；第二部分是实时计算开源技术选型与实践；第三部分是开源技术改造经验。
上图来自海尔早期的总体规划。从这个规划图可以看到，它主要包括两部分，一部分是离线计算，另一部分是实时计算。离线计算部分，海尔采用的技术主要是抽取离线的各种业务数据，放到 Hive，然后经过 Impala 进行数据分析。实时计算主要包括三部分工作，第一部分是某些关键业务数据的实时采集；第二部分是实时计算，而实时计算又包含两项工作，一个是实时数据分析，还有一个是核心业务的业务逻辑，比如实时订单、库存配送；第三部分是数据可视化分析的工具，这个工具的目的是解决实时计算和云计算的数据呈现问题，在搭建实时计算平台时，我们的原则是尽可能使用开源资源，而不用商业产品，因为不希望被商业厂家绑定。
常见的实时计算平台框架如上图。这个框架主要由三部分组成，首先是实时数据采集，然后采集的数据传输到实时计算框架，再将结果传输给数据可视化的呈现组织者，最后将一个个数据产品交给业务部门使用。实时计算在某些时候离不开离线计算的辅助。
接下来我们开始基于该框架进行技术选型。如今是技术爆炸的时代，实时计算所涉及的技术非常多。下图我简单列了一下，应该有将近 30 多种开源技术可供选择。
面对如此庞大的技术栈，如果我们一个个从头调研这些技术，将会耗费大量的时间，有时候免不了走一些弯路。
实时数据采集技术选型有三点要求，分别是：完整、低延时且不影响业务系统性能。完整即数据不能丢失；低延时实际上就是实时的概念；不影响业务系统性能，这一点其实会贯穿选型的始终，因为我们面对的生产系统最忌讳对它的应用系统产生任何可能的性能稳定性影响。
实时数据采集第一个要解决的问题就是如何获取数据。巧妇难为无米之炊，如果没有数据的话，一切都是空谈。对于数据的获取，业界通用的、最好的方案是埋点方案。埋点方案又包含两种，一种是代码埋点，另一种是可视化埋点。所谓代码埋点，即在代码或者业务系统里面插入探针类代码，当页面或控件被触发时进行采集数据。而另一种可视化埋点，指的是通过可视化的手段，提前选择要采集的控件操作，当控件操作发生时，再向后台发送数据。
比较这两个方案，较通用的是代码埋点。以购物为例，每一个商品购买页面，用户提交一个订单，那么通过可视化埋点可能就只能得到某时某刻某人提交了一个订单，而通过代码埋点可以得到更多信息，包括商品的名称、订单的金额、用户的各种属性等等。当然，代码埋点也有一个较为明显的问题，它对业务系统具有比较强的侵入性，需要在业务系统中写代码，并且人力、时间成本比较大。如果可以在系统开发时提前写好代码会好一些，而且代码埋点的技术难度其实并不大。
有了数据之后，我们开始进行日志收集。日志收集相关的开源技术框架中，第一个是 Flume。Flume 是一个开源且非常优秀的日志框架，我们可以把它理解成一个水池，它可以分成三部分，即 Source、Channel、Sink。其中 Channel 就是一个水池，Source 是它的进水管，Sink 是出水管。我们可以将进水管和出水管理解成，为它配置不同的管子支持不同的数据接入和 Sink 的不同目标。它们三个可以组成一个非常复杂的网络，由多个 Flume Agent 可以组成一道数据传输路由。
从架构上来说，Fluentd 与 Flume 非常相似，它带来了跨平台的能力，还有一个特性就是可以用 json 文件统一日志格式。
第三个框架也是非常优秀的日志收集框架，Logstash 更多的时候是跟 Elasticsearch 和 Impala 联合起来，用作运维方面的日志数据分析。
最后一个框架是 Scribe。Scribe 早在 2010 年就停止更新了，不过感兴趣的同学可以在网上查找一下资料进行了解。
实际上，前面三个开源框架都是非常优秀的。我们最终选择使用 Flume 技术框架进行日志收集。Flume 和 Fluentd 差不多，那么在相似的条件下，我们优先选择 Apache 开源产品。
Flume 监控官方推荐使用 Ganglia。Ganglia 监控主要是监控 Source、Channel、Sink 里面的 event 数。event 数实际上是一个个 log 日志，我们可以对比 Source 接收的 event 数和 Sink 已经处理的 event 数是否匹配，不匹配就报警。Channel 里也有 event 数，我们只需要关注它拥堵的 event 数，如果超过一定的阈值也需要报警。
日志数据获取包含几个过程：探针埋点、采集、收集、解析、入库。探针埋点实际上非常简单，它是纯粹的 JS 代码，我们会在业务系统上做一些 Cookie，通过 Cookie 获取到用户数据。JS 一旦被访问就会触发调用 Nginx 的 HTTP 服务。使用 Nginx 是因为它是一个性能很强大的 HTTP 服务器，它可以支持上万的并发量，而探针埋点对后台服务器的压力比较大，因此对并发要求较高。实际上 HTTP 只是记录访问，不会对后端进行任何操作，只会往 Nginx 里记录一条日志。我们对它采用不同的 JS 进行收集、过滤之后便可形成一个较为规范的数据格式，然后再发送到数据库里，或者说交到下一个环节的模块去处理。
上面介绍的是一个通用的数据获取方法，那么实际上还有新的一个问题，如果业务系统存在一些没有办法配合修改的情况，怎么办呢？这时我们只能从业务系统的数据库方面去想办法。
数据库里面最典型的是基于数据库 Change Data Capture 的方案。Change Data Capture，指的是变化数据捕获，我们称之为 CDC。CDC 有以下几种方法，实际上也可以归纳成两种，一种是侵入性的，另一种是非侵入性的。触发器、时间戳、全表比对这三类都属于侵入性 CDC 的获取方法，对业务系统的性能、数据库的性能会有一些影响，实时性也不是那么强。我们推荐使用日志对比的方案。日志对比，如 Oracle 日志、MySQL CDC 都支持，基于日志对比基本上对业务系统不会有任何影响。
针对各种数据库的 CDC 方案我们都做一下介绍。对于 Oracle，我们推荐使用 OGG 方案。OGG 方案是我们在所有的技术选型里面唯一使用的商业产品，因为相对来说 Oracle 还是比较封闭的，没有很好的开源解决方案。
我们之前也曾经尝试自己写代码，确实可以获取到 DML 和 DDL 语句的变化，但是捕获到了之后，会发现后面要做的事情非常多，你不得不去解决数据一致性的问题、顺序的问题、容错的问题等等，自己写的话需要一一考虑这些问题，实际上是得不偿失的，花费的时间也更长，因此我们直接使用 OGG 的商业产品。我们实际的生产库里有四个库，我们在备库上部署了 Oracle GoldenGate，还有 for big data 的 Oracle 组件部署在另外一个 OGG 上，由这两个 Oracle 的组件协作就可以捕获到 Oracle 的数据变化，并且投入到下一步的消息队列里面去。
所有的系统都需要进行监控，OGG 的监控也相当关键。OGG 的监控分为两种，一种是 GoldenGate Director，另一种是 GoldenGate Monitor，这两个功能实际上是类似的。但是 GoldenGate Monitor 提供的软件更为丰富，所以我们推荐使用 Monitor。
如今的大多数业务系统都使用了 MySQL 数据库。针对 MySQL，我们建议的方案就是大家首先会想到的 Canal，它是阿里巴巴的非常优秀的开源数据库，支持同步、增量订阅等功能。然而我们最终没有使用 Canal，因为需要为 Canal 客户端额外写一些代码，包括数据、消息的打包等，尤其是事务方面需要进一步考虑。我们查询了资料发现，GitHub 上有一个雅虎开源、可对外使用的方案 Yelp MySQLStreamer。Yelp 是美国最大的点评网站，它的开源组件能够直接获取数据的变化；它基于 MySQL 行的复制方式，而不是语义的方式；它可以直接将数据获取到下一步消息队列内，同时获取到 DDL、DML 的语义变化；基于该变化，可自动根据表进行打包，建立后面的消息队列的 topic。所以我们最终实现的方法是通过 MySQLstream。如果大家感兴趣，可以关注一下 Yelp MySQLStreamer。
Postgresql CDC 我们建议的方案是 BottledWater。Bottledwater 支持 Postgresql9.4 及以后版本，它是基于逻辑复制特性做出的框架，支持预写式日志，即 DLL 日志，类似于 Oracle 日志；它不兼容 9.4 以前的 Postgresql 版本；由于基于预写式日志，不会影响业务基本性能，并且也能保证事务一致性的输出。也就是说，只有在 commit 成功之后，消息才会发布到消息队列里面去；如果没有事务一致性的话，就可能导致还没 commit 成功，消息就发到消息队列了，最终导致后端数据不一致。最后一个特点是容错，在 Bottledwater 进程崩溃的情况下（包括机器故障、网络中断等），仍能保证数据不丢失。
通过各种方式获得实时数据后的下一步，我们需要引入消息队列，那么为什么要引入消息队列呢？Flume 只适合日志收集、传输以及数据的拦截。如果日志数据量非常多，Flume 可以拦截一部分数据，而消息队列做不到这一点，那么消息队列适合做什么呢？它适合消息持久化，因为消息队列具备消息的无限堆积能力，并且它持久化的速度非常快，这个特性可以结合流式计算的反压机制来解决突发的洪峰流量、突发数据激增，避免把后端的流式计算撑爆。消息队列还有一个优点是解耦。解耦指的是，当消息的发布方、接收方都不知道对方的存在，它可以在各个系统之间起到类似于消息总线的作用。我们最终采取的是 Flume 加消息队列的方式。
消息队列可选的方案非常多，甚至可以专门开一个专题讨论，这里大概罗列一下，包括 Kafka、JKafka、RocketMQ、RabbitMQ、ActiveMQ、Apollo。其中 JKafka 是 Kafka 的 Java 实现，并进行了一些功能方面的改进。ActiveMQ 应用也非常广泛，但在 6.0 版本之后基本上转到了 Apollo。
选择消息队列需从两个方面考虑：事务可靠性或吞吐量优先。因为我们的物流行业的数据量非常大，比如车位置信息等，所以这是一个大规模的数据传输。我们选择了吞吐量优先场景，使用的是 Kafka 消息队列。Kafka 在设计上没有遵循 JMS 规范，它的吞吐量在所有消息队列内是最大的。若采取此类单机消息的形式，一个 10 字节的消息基本上单机能达到每秒百万。实际上，阿里巴巴的 RocketMQ 也是借鉴了 Kafka 的设计，当然，它做了大量的定制工作，包括事务以及消息支持严格的顺序保证。而 ActiveMQ、RabbitMQ 大多数都是应用到企业应用系统的集成，因为它们支持的很多消息协议都是重量级的。
当我们部署 Kafka 时更加关注的是消息消费和数据变动的情况，因此我们需要一个监控管理界面。监控管理界面有四种选择，第一种 Kafka Web Console，第二种 Kafka Manager，第三种 KafkaOffsetMonitor，第四种是 Uber 开源的 Kafka 监控工具 Chaperone。
Kafka Web Console 比较全面；Kafka Manager 更多偏向 Kafka 本身的集群，甚至能进行 Kafka 分区的管理；KafkaOffsetMonitor 是一个轻量级的 Kafka 消息监控，；Uber 开源的 Chaperone 能实现闭合消息的监控，它是消息进一步分析的模式工具，因为它不是基于 KafkaOffsetMonitor 的偏移量，而是基于时间戳的，也就是说即使消息在 Kafka 内已经消费过了，Chaperone 也可以获取到并进一步分析。我们最终选择使用 KafkaOffsetMonitor，也许其他工具在一些方面会更强大，但是我们更关注监控分析，选一个轻量型的 Kafka Web 监控管理界面就能满足我们的要求。
流式计算可选方案非常多，因为我们之前已经部署了离线计算，所以在流式处理计算框架选取上，我们关注的是仅流处理框架，其中包含了 Storm、JStorm、Samza、Heron。在我们选型时业界已经存在很多流处理框架，但其中最常见的还是 Storm 框架，业界很多人一开始搭建实时计算平台是从 Storm 开始的。其实 Storm 很多时候在各种方面不如 JStrom 和 Heron，但 JStrom 和 Heron 在国内反而使用得比较少，更多的时候还是先基于 Storm 来搭建，在了解其中所有的环节、有了实践经验之后，再从 Strom 逐步升级到 JStrom 和 Heron，它们基本上可以做到无缝迁移。Samza 主要是能够更好、更方便地使用 Kafka，它是基于 Kafka 的一套工作方式来获取 Kafka 里面相应的队列变化。
如果从头开始搭建流式计算平台，也可以使用混合框架。混合框架主要包含两种：Spark 和 Flink。Flink 的实时性更好一些，但是在各方面的应用还是不如 Spark，因为 Spark 生态系统更加丰富，不仅支持离线计算，还支持实时计算、SQL 以及机器学习等。从综合能力上看，Spark 整体上要好于 Flink。
我们目前使用的是 Storm Trident，那么什么情况下应该使用 Storm Trident 呢？Trident 实际上是 Storm 更高一层的抽象，它除了提供一种简单应用的流计算 API 之外，还可以以 batch 的方式处理流数据，而且它能保证 exactly-once 方式，即确保只处理一次。这其实就是事务的概念，事务要么成功，要么就失败。基于这两个特性，Storm Trident 更适合有状态且需要保持状态的流处理使用；如果不需要保持状态，那么选择原先的 Storm 即可，因为一旦使用 batch，就会对性能造成影响，无法达到很好的实时性。
我们再看一下基于前面的技术选型最终得到的流式计算框架，上图为最常见的架构。我们把日志内的所有信息通过 Logback 获取到 Flume 里，然后在 Flume Sink 到 Kafka，再建立 Kafka Spout，最终组成了 Storm Topology。
整个流式处理到这是不是就结束了呢？实际上还需要考虑几点。第一点是借助实时计算数据分析的结果，得出结果需要给人查看，那么就需要一个方便的可视化工具；第二点是系统如何判断实时计算的结果有没有问题，业务人员一看数据即可得出结论，但是系统却需要有判断标准。这两点需要进一步考虑。
实时计算结果的正确性怎么保证？很多时候这一点都是需要优先考虑的，下面有两种方案。
第一种是以结果为导向，采用离线计算的方式来辅助实时计算。正如我之前 PPT 所提到的，在一些情况下，实时计算需要离线计算来辅助。这里将采用双线机制，获取到的数据给到实时计算的时候，同时以一小时的时间窗口发往离线计算，然后以一小时的时间窗口来进行计算。具体怎么计算呢？所有的实时结果与离线计算结果进行对比，采用对比方式的好处是，一旦发现问题就可以实时报警，并修正错误结果。为什么要采用双线机制，并且绝对不能把数据放到 Kafka 里？因为一旦通过这条线过来，如果在该点出了问题，或者 Kafka 本身出了问题，验证的方案就失效了。这套方案也存在一个缺点，就是发现问题可能不是那么及时，因为受限于离线计算一个小时的时间窗口，会有一个小时的滞后性。
第二种方法就是按照经验值来计算，其实就是借助模型预测。正如我们所知，数据分析中存在同比、环比的概念，很多时候，当我们拿到数据时会跟同比、环比的数据进行对比。这几个数据之间存在一些关联性，我们要根据各个不同的业务找出它们之间的关联性。打个比方，我们需要判断当前的 current 计算结果是否正确，就需要判断以下几个点：第一个点是当天的环比数据，以及昨天的、上个月的同比数据，然后基于这三个数据建立一个模型，再预测出当前的 current 值。那么这里面关键的难点是在什么地方呢？我们分析出的 current 值应该是一个范围，存在最高值和最低值，系统需要根据这个范围来报警，如果当前的 current 大于最高值或者低于最低值就报警。这一方法需要对业务非常熟悉才能进行分析预测，它的优点是能非常快地看到数据的异常激增或者丢失，实时性非常高。
在整个实时计算过程中，监控是非常重要的一个环节。我们选择 Ganglia 和 Nagios 平台来进行监控。为什么要将二者联合起来呢？因为 Ganglia 只适合做大量节点的监控，而不适合做报警；而 Nagios 可以实现报警功能，二者结合起来就能形成一个非常完美的监控系统。
海尔在开源技术改造的经验之一就是数据可视化方案的改造。实时计算得到的结果以什么样的方式被业务使用非常关键，我们的目标是为公司所有的数据分析业务提供数据洞察和展示工具，并提供给业务人员使用。海尔主要基于开源技术做数据可视化。
我们重点考查了以下三个可选方案，caravel、saiku、zeppelin。caravel 是非常优秀的数据可视化工具，支持各种类型分析，数据源、界面等都非常优秀，但是 caravel 主要基于 python 实现，而我们的团队主要使用的还是 Java，如果使用 caravel 后期还需要做一些代码上的更改，因此我们最终选择了 zeppelin。saiku 也是非常成熟的一个开源框架，但是它有商业版本，所以有些功能不会被放到开源版本内。
zeppelin 是 Apache 开源的数据可视化组件，上图来自 zeppelin 官网介绍。它支持数据的提取、发现、分析，包括数据可视化协作。从界面我们可以发现，它支持基本报表，包括分组、聚合、钻取、计算等，并且功能也较为强大。它的一个不足点是，仅从界面来看还远远满足不了设计报表的要求。
它更有优势的地方就是支持多语言，它对多语言的解析能力非常强大，几乎允许任何语言的接入。如上图所示，它所接入的语言非常多，不仅支持传统的关系型数据库，还支持大数据方面的技术，例如 Flink、Spark、HBase 等，各种几乎你能想到的语言它都已经接入了。对技术团队来说，如果我们使用较多的技术栈，借助 Zeppelin 就可以用一个统一的工具支持数据可视化的设计。那么支持这么多语言，它的源代码是在哪执行呢？打个比方，我们接入的是 Spark，那代码就是在 Spark 集群上执行，以此保证执行速度，因此它的响应速度完全取决于语言本身的速度。只要技术人员熟悉各个原生语言，可以在 Zeppelin 中直接写代码，得到计算结果之后就可以实现可视化的数据呈现。
Zeppelin 的技术架构是 web 应用程序的架构。如上图，最底层的 interpreter 是一个解释器，它以独立的 JVM 进程的方式运行，并执行各种语言的脚本获取数据，最终通过 Thrift 远程调用传给 Server 端，然后 Server 端再通过 websocket 的方式实时送达客户端。
那么我们对 Zeppelin 做了哪些改造呢？主要分为 Server 端改造和前端改造。对 Server 端的改造主要是使它能够支持 Echarts.js 数据，并将前台的逻辑计算改到后端，这样可以加快报表的处理速度、提高性能。前端的话，主要是把 nvd3.js 改成 Echarts, 因为 nvd3.js 大多数是国外使用，而国内大多数使用的是 Echarts。
我们还提供了一个强大易用的报表设计器。如图，报表设计器里面包括：数据执行区，可以执行任意语言，包括写 SQL 语句；字段控制区，可以对字段进行设置，比如添加计算字段，也就是建立一些表达式和基于现有的字段来设置成一个新字段；行列控制区，可以支持多层钻取、双轴异图、实时计算的同比、环比、度量维度排序等；标记卡控制区，标记卡有颜色、大小的标签，可利用颜色、大小对图进行系列的分类控制，比如按照某个颜色的维度进行报表的分组，对它进行颜色的分组会变成一个堆积图；再往下是辅助设置区，对此我们也做了大量工作，包括参数设置、图内筛选器、图表辅助线、图表预警等。以上整个 Zeppelin 系统就基本可以满足业务人员的使用要求。目前 Zeppelin 在行业内部是开源的，我们今年也计划推动公司将这个项目放到 GitHub 上开源。
最后总结一下我们目前的实时平台技术架构。实时数据采集方面，主要采用埋点的方法，利用 Flume+OGG，以及其他的一些组件；消息队列系统使用 Kafka；流式计算用 Storm；可视化分析利用 Zeppelin 工具呈现，综上所述即可打造一个完整的实时分析计算平台。
最后提一下 Solr6，为什么会采用 Solr6 呢？因为在进行报表分析的时候，有一些业务运营的系统数据需要查看，而且数据量可能非常大，如果直接用 Zeppelin 进行分析，报表速度可能太慢而无法忍受，这时就需要使用 Solr6，直接通过几个 Solr 集群来加速报表的数据采集。从这里我们也可以看出 Zeppelin 的强大，它可以直接连接 Solr6。
肖云，海尔电器互联网技术中心资深架构师，拥有多年互联网研发管理工作经验，历任方正电子新媒体开发总监、中投视讯平台研发总监，现就职于海尔电器互联网技术中心资深架构师，从事技术架构和大数据方面的研发管理工作。多年来专注于平台架构和大数据领域，喜欢研究开源项目，对于移动互联网平台能力架构和大数据平台建设方面，都有着深入的研究和实战经验。
今日荐文
点击下方图片即可阅读
分析海量视频中的违规内容，七牛如何构建弹性深度学习计算平台
深度学习零基础，如何在 9 周时间内，学会利用卷积网络实现对话 / 地主机器人，用 tensorflow 进行图片分类／人脸识别／模仿大师画作风格等实战技能？
我们邀请到资深大数据专家高扬和高级软件架构师卫峥两位老师开讲「深度学习，从入门到精通实战」课程，同学们将通过最浅显易懂的方式入门进阶深度学习。想了解更多，请扫海报二维码，点击阅读原文直接进入「课程链接」。
相关内容
微信扫码咨询专知VIP会员