Spark Streaming + Elasticsearch构建App异常监控平台

标签: spark streaming elasticsearch | 发表时间:2016-11-04 14:38 | 作者:美团点评技术团队
出处:http://tech.meituan.com/

本文已发表在《程序员》杂志2016年10月期。

如果在使用App时遇到闪退,你可能会选择卸载App、到应用商店怒斥开发者等方式来表达不满。但开发者也同样感到头疼,因为崩溃可能意味着用户流失、营收下滑。为了降低崩溃率,进而提升App质量,App开发团队需要实时地监控App异常。一旦发现严重问题,及时进行热修复,从而把损失降到最低。App异常监控平台,就是将这个方法服务化。

低成本

小型创业团队一般会选择第三方平台提供的异常监控服务。但中型以上规模的团队,往往会因为不想把核心数据共享给第三方平台,而选择独立开发。造轮子,首先要考虑的就是成本问题。我们选择了站在开源巨人的肩膀上,如图1所示。

 图1 数据流向示意图

Spark Streaming

每天来自客户端和服务器的大量异常信息,会源源不断的上报到异常平台的Kafka中,因此我们面临的是一个大规模流式数据处理问题。美团点评数据平台提供了Storm和Spark Streaming两种流式计算解决方案。我们主要考虑到团队之前在Spark批处理方面有较多积累,使用Spark Streaming成本较低,就选择了后者。

Elasticsearch

Elasticsearch(后文简称ES),是一个开源搜索引擎。不过在监控平台中,我们是当做“数据库”来使用的。为了降低展示层的接入成本,我们还使用了另一个开源项目ES SQL提供类SQL查询。ES的运维成本,相对 SQL on HBase方案也要低很多。整个项目开发只用了不到700行代码,开发维护成本还是非常低的。那如此“简单”的系统,可用性可以保证吗?

高可用

Spark Streaming + Kafka的组合,提供了“Exactly Once”保证:异常数据经过流式处理后,保证结果数据中(注:并不能保证处理过程中),每条异常最多出现一次,且最少出现一次。保证Exactly Once是实现24/7的高可用服务最困难的地方。在实际生产中会出现很多情况,对Exactly Once的保证提出挑战:

异常重启

Spark提供了Checkpoint功能,可以让程序再次启动时,从上一次异常退出的位置,重新开始计算。这就保证了即使发生异常情况,也可以实现每条数据至少写一次HDFS。再覆写相同的HDFS文件就保证了Exactly Once(注:并不是所有业务场景都允许覆写)。写ES的结果也一样可以保证Exactly Once。你可以把ES的索引,就当成HDFS文件一样来用:新建、删除、移动、覆写。
作为一个24/7运行的程序,在实际生产中,异常是很常见的,需要有这样的容错机制。但是否遇到所有异常,都要立刻挂掉再重启呢?显然不是,甚至在一些场景下,你即使重启了,还是会继续挂掉。我们的解决思路是:尽可能把异常包住,让异常发生时,暂时不影响服务。

 图 2 作业异常重启架构图

如图2所示,包住异常,并不意味可以忽略它,必须把异常收集到Spark Driver端,接入监控(报警)系统,人工判断问题的严重性,确定修复的优先级。
为了更好地掌控Spark Streaming服务的状态,我们还单独开发了一个作业调度(重启)工具。美团点评数据平台安全认证的有效期是7天,一般离线的批处理作业很少会运行超过这个时间,但Spark Streaming作业就不同了,它需要一直保持运行,所以作业只要超过7天就会出现异常。因为没有找到优雅的解决方案,只好粗暴地利用调度工具,每周重启刷新安全认证,来保证服务的稳定。

升级重导

Spark提供了2种读取Kafka的模式:“Receiver-based Approach”和“Direct Approach”。使用Receiver模式,在极端情况下会出现Receiver OOM问题。
使用Direct模式可以避免这个问题。我们使用的就是这种Low-level模式,但在一些情况下需要我们自己维护Kafka Offset:
升级代码:开启Checkpoint后,如果想改动代码,需要清空之前的Checkpoint目录后再启动,否则改动可能不会生效。但当这样做了之后,就会发现另一个问题——程序“忘记”上次读到了哪个位置,因为存储在Checkpoint中的Offset信息也一同被清空了。这种情况下,需要自己用ZooKeeper维护Kafka的Offset。
重导数据:重导数据的场景也是,当希望从之前的某一个时间点开始重新开始计算的时候,显然也需要自己维护时间和Offset的映射关系。
自己维护Offset的成本并不高,所以看起来Checkpoint功能很鸡肋。其实可以有一些特殊用法的,例如,因为Python不需要编译,所以如果使用的是PySpark,可以把主要业务逻辑写在提交脚本的外边,再使用Import调用。这样升级主要业务逻辑代码时,只要重启一下程序即可。网上有不少团队分享过升级代码的“黑科技”,这里不再展开。
实现24/7监控服务,我们不仅要解决纯稳定性问题,还要解决延迟问题。

低延迟

App异常监控,需要保证数据延迟在分钟级。
虽然Spark Streaming有着强大的分布式计算能力,但要满足用户角度的低延迟,可不是单纯的能计算完这么简单。

输入问题

iOS App崩溃时,会生成Crash Log,但其内容是一堆十六进制的内存地址,对开发者来说就是“天书”。只有经过“符号化”的Crash Log,开发者才能看懂。因为符号化需要在Mac环境下进行,而我们的Mac集群资源有限,不能符号化全部Crash Log。即使做了去重等优化,符号化后的数据流还是有延迟。每条异常信息中,包含N维数据,如果不做符号化只能拿到其中的M维。

 图 3 双延迟乱序数据流融合示意图

如图3所示,我们将数据源分为符号化数据流、未符号化数据流,可以看出两个数据流的相对延迟时间T较稳定。如果直接使用符号化后的数据流,那么全部N维数据都会延迟时间T。为了降低用户角度的延迟,我们根据经验加大了时间窗口:先存储未符号化的M维数据,等到拿到对应的符号化数据后,再覆写全部N维数据,这样就只有N-M维数据延迟时间T了。

输出问题

如果Spark Streaming计算结果只是写入HDFS,很难遇到什么性能问题。但你如果想写入ES,问题就来了。因为ES的写入速度大概是每秒1万行,只靠增加Spark Streaming的计算能力,很难突破这个瓶颈。
异常数据源的特点是数据量的波峰波谷相差巨大。由于我们使用了 Direct 模式,不会因为数据量暴涨而挂掉,但这样的“稳定”从用户角度看没有任何意义:短时间内,数据延迟会越来越大,暴增后新出现的异常无法及时报出来。为了解决这个问题,我们制定了一套服务降级方案。

 服务降级方案示意图

如图4所示,我们根据写ES的实际瓶颈K,对每个周期处理的全部数据N使用水塘抽样(比例K/N),保证始终不超过瓶颈。并在空闲时刻使用Spark批处理,将N-K部分从HDFS补写到ES。既然写ES这么慢,那我们为什么还要用ES呢?

高性能

开发者需要在监控平台上分析异常。实际分析场景可以抽象描述为:“实时 秒级 明细 聚合” 数据查询。
我们团队在使用的OLAP解决方案可以分为4种,它们各有各的优势:

  • SQL on HBase方案,例如:Phoenix、Kylin。我们团队从2015年Q1开始,陆续在SEM、SEO生产环境中使用Phoenix、Kylin至今。Phoenix算是一个“全能选手”,但更适合业务模式较固定的场景;Kylin是一个很不错的OLAP产品,但它的问题是不能很好支持实时查询和明细查询,因为它需要离线预聚合。另外,基于其他NoSQL的方案,基本大同小异,如果选择HBase,建议团队在HBase运维方面有一定积累。
  • SQL on HDFS方案,例如:Presto、Spark SQL。这两个产品,因为只能做到亚秒级查询,我们平时多用在数据挖掘的场景中。
  • 时序数据库方案,例如:Druid、OpenTSDB。OpenTSDB是我们旧版App异常监控系统使用过的方案,更适合做系统指标监控。
  • 搜索引擎方案,代表项目有ES。相对上面的3种方案,基于倒排索引的ES非常适合异常分析的场景,可以满足:实时、秒级、明细、聚合,全部4种需求。

ES在实际使用中的表现如何呢?

明细查询

支持明显查询,算是ES的主要特色,但因为是基于倒排索引的,明细查询的结果最多只能取到10000条。在异常分析中,使用明细查询的场景,其实就是追查异常Case,根据条件返回前100条就能满足需求了。例如:已知某设备出现了Crash,直接搜索这个设备的DeviceId就可以看到这个设备最近的异常数据。我们在生产环境中做到了95%的明细查询场景1秒内返回。

聚合查询

面对爆炸的异常信息,一味追求全是不现实,也是没必要的。开发者需要能快速发现关键问题。
因此平台需要支持多维度聚合查询,例如按模块版本机型城市等分类聚合,如图5所示。

 图 5 聚合查询页面截图

不用做优化,ES聚合查询的性能就已经可以满足需求。因此,我们只做了一些小的使用改进,例如:很多异常数据在各个维度的值都是相同的,做预聚合可以提高一些场景下的查询速度。开发者更关心最近48小时发生的异常,分离冷热数据,自动清理历史数据也有助于提升性能。最终在生产环境中,做到了90%的聚合查询场景1秒内返回。

可扩展

异常平台不止要监控App Crash,还要监控服务端的异常、性能等。不同业务的数据维度是不同的,相同业务的数据维度也会不断的变化,如果每次新增业务或维度都需要修改代码,那整套系统的升级维护成本就会很高。

维度

为了增强平台的可扩展性,我们做了全平台联动的动态维度扩展:如果App开发人员在日志中新增了一个“城市”维度,那么他不需要联系监控平台做项目排期,立刻就可以在平台中查询“城市”维度的聚合数据。只需要制定好数据收集、数据处理、数据展示之间的交互协议,做到动态维度扩展就很轻松了。需要注意的是,ES中需要聚合的维度,Index要设置为“not_analyzed”。
想要支持动态字段扩展,还要使用动态模板,样例如下:

  {
    "mappings": {
        "es_type_name": {
            "dynamic_templates": [
                {
                    "template_1": {
                        "match": "*log*",
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "string"
                        }
                    }
                },
                {
                    "template_2": {
                        "match": "*",
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "string",
                            "index": "not_analyzed"
                        }
                    }
                }
            ]
        }
    }
}

资源

美团点评数据平台提供了Kafka、Spark、ES的集群,整套技术栈在资源上也是分布式可扩展的。
线上集群使用的版本:

  • kafka-0.8.2.0
  • spark-1.5.2
  • elasticsearch-2.1.1

相关 [spark streaming elasticsearch] 推荐:

Spark Streaming + Elasticsearch构建App异常监控平台

- - 美团点评技术团队
本文已发表在《程序员》杂志2016年10月期. 如果在使用App时遇到闪退,你可能会选择卸载App、到应用商店怒斥开发者等方式来表达不满. 但开发者也同样感到头疼,因为崩溃可能意味着用户流失、营收下滑. 为了降低崩溃率,进而提升App质量,App开发团队需要实时地监控App异常. 一旦发现严重问题,及时进行热修复,从而把损失降到最低.

[原]Spark Streaming原理简析

- - 张包峰的博客
StreamingContext实例化的时候,需要传入一个 SparkContext,然后指定要连接的 spark matser url,即连接一个 spark engine,用于获得executor. 实例化之后,首先,要指定一个接收数据的方式,如. 这样从socket接收文本数据. 这个步骤返回的是一个 ReceiverInputDStream的实现,内含 Receiver,可接收数据并转化为RDD放内存里.

Spark Streaming 自定义接收器

- - zzm
Spark Streaming可以从任意数据源接受流数据,而不仅仅是那些内置支持的数据源(如Flume、kafka等). 这就要求开发人员实现一个接收器(recevier),用于接收来自有关数据源的数据. 本篇手册以一个自定义的接收器(recevier)实现和其在spark streaming中的应为为主线进行讲解.

Kafka+Spark Streaming+Redis实时计算整合实践

- - 简单之美
基于Spark通用计算平台,可以很好地扩展各种计算类型的应用,尤其是Spark提供了内建的计算库支持,像Spark Streaming、Spark SQL、MLlib、GraphX,这些内建库都提供了高级抽象,可以用非常简洁的代码实现复杂的计算逻辑、这也得益于Scala编程语言的简洁性. 这里,我们基于1.3.0版本的Spark搭建了计算平台,实现基于Spark Streaming的实时计算.

Spark Streaming+kafka订单实时统计实现

- - CSDN博客推荐文章
前几篇文章我们分别学习Spark RDD和PairRDD编程,本文小编将通过简单实例来加深对RDD的理解. 开发环境:window7+eclipse+jdk1.7. 部署环境:linux+zookeeper+kafka+hadoop+spark. 本实例开发之前,默认已搭好了开发环境和部署环境,如果未搭建,可以参考本人相关大数据开发搭建博客.

Spark Streaming vs. Kafka Stream 哪个更适合你

- - IT瘾-bigdata
作者:Mahesh Chand Kandpal. 译者注:本文介绍了两大常用的流式处理框架,Spark Streaming和Kafka Stream,并对他们各自的特点做了详细说明,以帮助读者在不同的场景下对框架进行选择. 流式处理的需求每天都在增加,仅仅对大量的数据进行处理是不够的. 数据必须快速地得到处理,以便企业能够实时地对不断变化的业务环境做出反应.

Spark生态圈之——Elasticsearch与Solr

- - 企业架构 - ITeye博客
Elasticsearch简介. Elasticsearch是一个实时的分布式搜索和分析引擎. 它可以帮助你用前所未有的速度去处理大规模数据. 它可以用于全文搜索,结构化搜索以及分析,当然你也可以将这三者进行组合. Elasticsearch是一个建立在全文搜索引擎 Apache Lucene™ 基础上的搜索引擎,可以说Lucene是当今最先进,最高效的全功能开源搜索引擎框架.

Spark 实战, 第 2 部分:使用 Kafka 和 Spark Streaming 构建实时数据处理系统

- -
Spark 实战, 第 2 部分:使用 Kafka 和 Spark Streaming 构建实时数据处理系统. 2015 年 7 月 27 日发布. 在很多领域,如股市走向分析, 气象数据测控,网站用户行为分析等,由于数据产生快,实时性强,数据量大,所以很难统一采集并入库存储后再做处理,这便导致传统的数据处理架构不能满足需要.

Hadoop Streaming 编程

- - 学着站在巨人的肩膀上
Hadoop Streaming是Hadoop提供的一个编程工具,它允许用户使用任何可执行文件或者脚本文件作为Mapper和Reducer,例如:. 采用shell脚本语言中的一些命令作为mapper和reducer(cat作为mapper,wc作为reducer). 本文安排如下,第二节介绍Hadoop Streaming的原理,第三节介绍Hadoop Streaming的使用方法,第四节介绍Hadoop Streaming的程序编写方法,在这一节中,用C++、C、shell脚本 和python实现了WordCount作业,第五节总结了常见的问题.

Spark概览

- - 简单文本
Spark具有先进的DAG执行引擎,支持cyclic data flow和内存计算. 因此,它的运行速度,在内存中是Hadoop MapReduce的100倍,在磁盘中是10倍. 这样的性能指标,真的让人心动啊. Spark的API更为简单,提供了80个High Level的操作,可以很好地支持并行应用.