前言
 
  美团是数据驱动的互联网服务,用户每天在美团上的点击、浏览、下单支付行为都会产生海量的日志,这些日志数据将被汇总处理、分析、挖掘与学习,为美团的各种推荐、搜索系统甚至公司战略目标制定提供数据支持。大数据处理渗透到了美团各业务线的各种应用场景,选择合适、高效的数据处理引擎能够大大提高数据生产的效率,进而间接或直接提升相关团队的工作效率。
 
  美团最初的数据处理以HiveSQL为主,底层计算引擎为MapReduce,部分相对复杂的业务会由工程师编写MapReduce程序实现。随着业务的发展,单纯的HiveSQL查询或者MapReduce程序已经越来越难以满足数据处理和分析的需求。
 
  一方面,MapReduce计算模型对多轮迭代的DAG作业支持不给力,每轮迭代都需要将数据落盘,极大地影响了作业执行效率,另外只提供Map和Reduce这两种计算因子,使得用户在实现迭代式计算(比如:机器学习算法)时成本高且效率低。
 
  另一方面,在数据仓库的按天生产中,由于某些原始日志是半结构化或者非结构化数据,因此,对其进行清洗和转换操作时,需要结合SQL查询以及复杂的过程式逻辑处理,这部分工作之前是由HiveSQL结合Python脚本来完成。这种方式存在效率问题,当数据量比较大的时候,流程的运行时间较长,这些ETL流程通常处于比较上游的位置,会直接影响到一系列下游的完成时间以及各种重要数据报表的生成。
 
  基于以上原因,美团在2014年的时候引入了Spark。为了充分利用现有Hadoop集群的资源,我们采用了SparkonYarn模式,所有的Sparkapp以及MapReduce作业会通过Yarn统一调度执行。Spark在美团数据平台架构中的位置如图所示:
 
 
  经过近两年的推广和发展,从最开始只有少数团队尝试用Spark解决数据处理、机器学习等问题,到现在已经覆盖了美团各大业务线的各种应用场景。从上游的ETL生产,到下游的SQL查询分析以及机器学习等,Spark正在逐步替代MapReduce作业,成为美团大数据处理的主流计算引擎。目前美团Hadoop集群用户每天提交的Spark作业数和MapReduce作业数比例为4:1,对于一些上游的HiveETL流程,迁移到Spark之后,在相同的资源使用情况下,作业执行速度提升了十倍,极大地提升了业务方的生产效率。
 
  下面我们将介绍Spark在美团的实践,包括我们基于Spark所做的平台化工作以及Spark在生产环境下的应用案例。其中包含Zeppelin结合的交互式开发平台,也有使用Spark任务完成的ETL数据转换工具,数据挖掘组基于Spark开发了特征平台和数据挖掘平台,另外还有基于Spark的交互式用户行为分析系统以及在SEM投放服务中的应用,以下是详细介绍。
 
  Spark交互式开发平台
 
  在推广如何使用Spark的过程中,我们总结了用户开发应用的主要需求:
 
  数据调研:在正式开发程序之前,首先需要认识待处理的业务数据,包括:数据格式,类型(若以表结构存储则对应到字段类型)、存储方式、有无脏数据,甚至分析根据业务逻辑实现是否可能存在数据倾斜等等。这个需求十分基础且重要,只有对数据有充分的掌控,才能写出高效的Spark代码;
 
  代码调试:业务的编码实现很难保证一蹴而就,可能需要不断地调试;如果每次少量的修改,测试代码都需要经过编译、打包、提交线上,会对用户的开发效率影响是非常大的;
 
  联合开发:对于一整个业务的实现,一般会有多方的协作,这时候需要能有一个方便的代码和执行结果共享的途径,用于分享各自的想法和试验结论。
 
  基于这些需求,我们调研了现有的开源系统,最终选择了Apache的孵化项目Zeppelin,将其作为基于Spark的交互式开发平台。Zeppelin整合了Spark,Markdown,Shell,Angular等引擎,集成了数据分析和可视化等功能。
 
 
  我们在原生的Zeppelin上增加了用户登陆认证、用户行为日志审计、权限管理以及执行Spark作业资源隔离,打造了一个美团的Spark的交互式开发平台,不同的用户可以在该平台上调研数据、调试程序、共享代码和结论。
 
  集成在Zeppelin的Spark提供了三种解释器:Spark、Pyspark、SQL,分别适用于编写Scala、Python、SQL代码。对于上述的数据调研需求,无论是程序设计之初,还是编码实现过程中,当需要检索数据信息时,通过Zeppelin提供的SQL接口可以很便利的获取到分析结果;另外,Zeppelin中Scala和Python解释器自身的交互式特性满足了用户对Spark和Pyspark分步调试的需求,同时由于Zeppelin可以直接连接线上集群,因此可以满足用户对线上数据的读写处理请求;最后,Zeppelin使用WebSocket通信,用户只需要简单地发送要分享内容所在的http链接,所有接受者就可以同步感知代码修改,运行结果等,实现多个开发者协同工作。
 
  Spark作业ETL模板
 
  除了提供平台化的工具以外,我们也会从其他方面来提高用户的开发效率,比如将类似的需求进行封装,提供一个统一的ETL模板,让用户可以很方便的使用Spark实现业务需求。
 
  美团目前的数据生产主体是通过ETL将原始的日志通过清洗、转换等步骤后加载到Hive表中。而很多线上业务需要将Hive表里面的数据以一定的规则组成键值对,导入到Tair中,用于上层应用快速访问。其中大部分的需求逻辑相同,即把Hive表中几个指定字段的值按一定的规则拼接成key值,另外几个字段的值以json字符串的形式作为value值,最后将得到的对写入Tair。
 
 
  由于Hive表中的数据量一般较大,使用单机程序读取数据和写入Tair效率比较低,因此部分业务方决定使用Spark来实现这套逻辑。最初由业务方的工程师各自用Spark程序实现从Hive读数据,写入到Tair中(以下简称hive2Tair流程),这种情况下存在如下问题:
 
  每个业务方都要自己实现一套逻辑类似的流程,产生大量重复的开发工作;
 
  由于Spark是分布式的计算引擎,因此代码实现和参数设置不当很容易对Tair集群造成巨大压力,影响Tair的正常服务。
 
  基于以上原因,我们开发了Spark版的hive2Tair流程,并将其封装成一个标准的ETL模板,其格式和内容如下所示:
 
 
  source用于指定Hive表源数据,target指定目标Tair的库和表,这两个参数可以用于调度系统解析该ETL的上下游依赖关系,从而很方便地加入到现有的ETL生产体系中。
 
  有了这个模板,用户只需要填写一些基本的信息(包括Hive表来源,组成key的字段列表,组成value的字段列表,目标Tair集群)即可生成一个hive2Tair的ETL流程。整个流程生成过程不需要任何Spark基础,也不需要做任何的代码开发,极大地降低了用户的使用门槛,避免了重复开发,提高了开发效率。该流程执行时会自动生成一个Spark作业,以相对保守的参数运行:默认开启动态资源分配,每个Executor核数为2,内存2GB,最大Executor数设置为100。如果对于性能有很高的要求,并且申请的Tair集群比较大,那么可以使用一些调优参数来提升写入的性能。目前我们仅对用户暴露了设置Executor数量以及每个Executor内存的接口,并且设置了一个相对安全的最大值规定,避免由于参数设置不合理给Hadoop集群以及Tair集群造成异常压力。
 
  基于Spark的用户特征平台
 
  在没有特征平台之前,各个数据挖掘人员按照各自项目的需求提取用户特征数据,主要是通过美团的ETL调度平台按月/天来完成数据的提取。
 
  但从用户特征来看,其实会有很多的重复工作,不同的项目需要的用户特征其实有很多是一样的,为了减少冗余的提取工作,也为了节省计算资源,建立特征平台的需求随之诞生,特征平台只需要聚合各个开发人员已经提取的特征数据,并提供给其他人使用。特征平台主要使用Spark的批处理功能来完成数据的提取和聚合。
 
  开发人员提取特征主要还是通过ETL来完成,有些数据使用Spark来处理,比如用户搜索关键词的统计。
 
  开发人员提供的特征数据,需要按照平台提供的配置文件格式添加到特征库,比如在图团购的配置文件中,团购业务中有一个用户24小时时段支付的次数特征,输入就是一个生成好的特征表,开发人员通过测试验证无误之后,即完成了数据上线;另外对于有些特征,只需要从现有的表中提取部分特征数据,开发人员也只需要简单的配置即可完成。
 
 
  在图中,我们可以看到特征聚合分两层,第一层是各个业务数据内部聚合,比如团购的数据配置文件中会有很多的团购特征、购买、浏览等分散在不同的表中,每个业务都会有独立的Spark任务来完成聚合,构成一个用户团购特征表;特征聚合是一个典型的join任务,对比MapReduce性能提升了10倍左右。第二层是把各个业务表数据再进行一次聚合,生成最终的用户特征数据表。
 
  特征库中的特征是可视化的,我们在聚合特征时就会统计特征覆盖的人数,特征的最大最小数值等,然后同步到RDB,这样管理人员和开发者都能通过可视化来直观地了解特征。另外,我们还提供特征监测和告警,使用最近7天的特征统计数据,对比各个特征昨天和今天的覆盖人数,是增多了还是减少了,比如性别为女这个特征的覆盖人数,如果发现今天的覆盖人数比昨天低了1%(比如昨天6亿用户,女性2亿,那么人数降低了1%*2亿=2百万)突然减少2万女性用户说明数据出现了极大的异常,何况网站的用户数每天都是增长的。这些异常都会通过邮件发送到平台和特征提取的相关人。