Spark大数据分析实战.docx

《Spark大数据分析实战.docx》由会员分享，可在线阅读，更多相关《Spark大数据分析实战.docx（302页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、971 SNAP符号社交网络Epinions数据集 972基于Spark的链路预测算法9.8本章小结第10章基于Spark的大规模新闻主题分析主题模型简介10.1 主题模型LDA1021 LDA模型介绍1022 LDA的训I练算法Spark中的LDA模型10.1.1 MLlib 对 LDA 的支持Spark中LDA模型训练示例10.2 案例：Newsgroups 新闻的主题分析Newsgroups 数据集介绍1042交叉验证估计新闻的主题个数1043基于主题模型的文本聚类算法10.4.4 基于主题模型的文本分类算法本章小结第11章构建分布式的搜索引擎当数据量每天以10GB、1OOGB增长的时候

2、，单机处理能力已经不能满足需求。此时就需 要增加系统的扩展性，用大数据分析和并行计算来解决。在Spark出现之前，海量数据 存储和海量日志分析都是基于Hadoop、Hive等数据分析系统的。Spark的出现，使得全 栈数据分析更加容易。并且，Spark非常适合构建多范式日志分析流水线。我们将介绍如 何使用Spark构建日志分析流水线。4.2 日志分析指标下面将介绍常用网站的运营数据分析指标。在数据越来越重要的趋势下，数据化运营已经 提上互联网公司的日程，如果监控网站或应用的状况时发现瓶颈问题，我们需要针对网站 或应用相关指标进行统计和分析得出的。随着移动互联网的发展，越来越多的移动数据分 析公

3、司与工具也不断涌现，其中代表性的为友盟、TalkingData等，为公司提供数据化运 营支持。网站运行日志分析常用指标如下： PV (PageView)：网站页面访问数，也称作网站流量。 UV (Unique Visitor)：页面IP的访问量统计，访问用户数，即独立IP。 PVPU (Page View Per User)：平均每位用户访问页面数。.漏斗模型与转化率：漏斗模型指的是多个不同的事件按照一定依赖顺序依次触发的流程 中的转化模型。用户通常会对应用中的一些关键路径进行分析。比如注册流程、购物流程、 交易流程等。以电商应用的购物流程为例： 1浏览商品页-2放入购物车t3生成订单-4支付

4、订单-5完成交易我们可以根据这些关键路径来计算每一步的转化率。转化率指的是完成当前事件的用户 中触发下一个依赖事件的用户所占比例。留存率：用户在某段时间内开始使用应用，经过一段时间后，仍然继续使用这个应用的 用户被认作是留存。这部分用户占开始新增用户的比例即是留存率。用户属性：用户的基本属性和行为特征，将用户打标签，帮助产品进一步的营销与推荐。最终希望通过一个仪表盘展示出整个网站的统计指标信息，如图4-1所示。Wekome!Visitor MapOAK RANCE: 2015 04 27wnr at aw* vnotooVisitors in Real-time(MikIjacf 74 hou

5、rsVisitor BrowserKeywords30 mintitosBftOWSCRKEYWtMDItMwtey 2S E 7015 03:16:70 fl mr 6 Chromea Keyword not tfeflfwd1000UNIQUE VIS11URSYou 3n clos this wldgt by cHckln4 on the X koc at the top of the widget.visitsACI1UKS1357?B40图4-1日志统计效果图4.3 Lamda 架构日志分析中既有离线大规模分析的需求，又有实时性的需求，这就可以通过采用Lamda 架构构建日志分析流水

6、线。1.Lamda架构简介Lambda架构的目的是为大数据分析应用程序提供一个低响应延迟的组合数据传输环境。Lambda系统架构定义了一套明确的架构原则，它为建立一套强大的和可扩展的数据系统 定义了架构范式。在Lamda架构中，被读取的数据是不可变的，在并行处理过程中数据会依次进入流处理系统和批处理系统，同时进行实时处理和离线数据分析。在查询时，当 这两者都返回结果后，才算是完成一次完整的查询。从逻辑上看，传输过程发生了两次， 一次是在批处理中，一次是在流处理中。Lamda架构并不限定其中的具体系统，要根据实际情况进行调整优化。大数据的系统选 型具体可以有很多的组合变化。例如可以将图4-2中的

7、Kafka、Storm. Hadoop等换成其 他类似的系统，例如Spark Streaming、Spark等，惯常的做法是使用两个数据库来存储 数据输出表，一个存储实时表，响应实时查询需求，另外一个存储批处理表，返回离线计 算结果。All data(uuols)女 PO2SRealtimeviewPrecomputeviewsBatch viewBatch viewServing layerStreamprocessing图4-2 Lamda数据分析架构它是由三层组成：批处理层、服务层和速度层。批处理层：Hadoop Spark Tez等都可以作为批处理层的处理工具，HDFS HBase 等

8、都可以作为数据持久化系统。服务层：用于加载和实现数据库中的批处理视图，以便用户能够查询，不一定需要随 机写，但是支持批更新和随机读，例如采用ElephantDB Voldemorto快速处理层：主要处理新数据和服务层更新造成的高延迟补偿，利用流处理系统（如 Storm S4、Spark Streaming）和随机读写数据存储库来计算实时视图（HBase）。批处 理和服务层定期处理和转换实时视图为批处理视图。为了获得一个完整结果，批处理和实时视图都必须被同时查询和融合（实时代表新数据）。下面借鉴Lamda架构，设计整个数据分析流水线架构，如图4.3所示。KafkaHadoop (HDFS)- S

9、eries 1 Series! Series?图4-3日志分析流水线整体架构图本例中实时日志分析流水线大致按以下步骤操作。数据采集：采用Flume NG进行数据采集。数据汇总与转发：通过Flume将数据转发汇总到实时消息系统Kafkao数据处理：采用Spark Streaming进行实时数据处理。结果呈现：采用Flask作为可视化呈现工具进行结果呈现。离线日志分析流水线大致按以下步骤操作。数据存储：通过Flume将数据转储到HDFSo数据处理：通过Spark SQL进行数据预处理。结果呈现：结果汇总存储到MySQL最后通过Flask进行结果呈现。4.4构建日志分析数据流水线后续的章节将介绍日志

10、数据采集、日志数据汇总、日志实时分析、日志离线分析及可视化, 来构建数据分析流水线。4.4.1 用Flume进行日志采集Web日志由Web服务器产生，生产环境的服务器可能是Nginx、Apache Tomcat IIS 等。例如，可以将Tomcat的日志收集到指定的目录，Tomcat安装在/opt/tomcat,日志存放 在var/log/data。其他服务器（如Apache、Nginx、HS等），用户可以根据相应服务器 的默认目录进行相关配置。1 .Flume 简介Flume是Cloudera开发的日志收集系统，具有分布式、高可用等特点，为大数据日志采 集、汇总聚合和转储传输提供了支持。为了

11、保证Flume的扩展性和灵活性，在日志系统 中定制各类数据发送方及数据接收方。同时Flume提供对数据进行简单处理，并写各种 数据到接受方的能力。Flume的核心是把数据从数据源收集过来，再送到数据接收方。为了保证送达成功，在送 到目的地之前，会先缓存数据，待数据真正到达目的地后，删除自己缓存的数据。Flume传输的数据的基本单位是事件（Event）,如果是文本文件，通常是一行记录，这 也是事务的基本单位。事件（Event）从源（Source）传输到通道（Channel）,再到数据 输出槽（Sink）,本身为一个比特（byte）数组，并可携带消息头（headers）信息。Flume运行的核心是

12、Agent。它是一个完整的数据收集工具，含有三个核心组件，分别是 Source Channel、Sinko通过这些组件，Event可以从一个地方流向另一个地方，如图 4-4所不。图4-4 Flume架构Flume核心组件如下。Source可以接收外部源发送过来的数据。不同的Source,可以接受不同的数据格式。比 如有目录池(Spooling Directory)数据源，可以监控指定文件夹中的新文件变化，如果 目录中有文件产生，就会立刻读取其内容。Channel是一个存储地，接收Source的输出，直到有Sink消费掉Channel中的数据。Channel中的数据直到进入到下一个Channel

13、中或者进入终端才会被删除。当Sink写入 失败后，可以自动重启，不会造成数据丢失，因此很可靠。Sink会消费Channel中的数据，然后送给外部源或者其他Source。如数据可以写入到 HDFS 或者 HBase 中。Flume允许多个Agent连在一起，形成前后相连的多级数据传输通道。2 .Flume安装与配置(1)安装 Flume1)安装JDK。2)安装 Flume。http:/ # tar xvzf apache-flume-1.5.0-bin.tar.gz # mv apache-flume-1.5.0-bin apache-flume-1.5.0 # In -s apache-flu

14、me-1.5.0 flume3)环境变量设置。# vim /etc/profile export JAVA_HOME=/usr/local/jdk exportCLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar exportPATH=$ PATH: $ J AVA_H 0 M E/bin export FLUME_HOME=/usr/local/flume export FLUME_CONF_DIR=$FLUME_HOME/conf export PATH=.:$PATH:$FLUME_HOME/bin # source /et

15、c/profile(2)创建Agent配置文件将数据输出到HDFS这需要修改flume.conf中的配置，具体如下：al.sources = rl al.sinks = kl al.channels = cl # 描述和配置 source # 第 1 步：配置数据源 al.sources.rl.type = exec al.sources.rl.channels = cl # 配置需要监控的日志输出目录 mand = tail -F /var/log/data # Describe the sink # 第 2 步：配置数据输 出 al.sinks.kl.type=hdfs al.sinks.

16、kl.channel=cl al.sinks.kl.hdfs.useLocalTimeStamp=true al.sinks.kl.hdfs.path=hdfs:/192.168.11.177:9000/flume/events/%Y/%m/%d/%H/%M al.sinks.kl.hdfs.filePrefix=cmcc al.sinks.kl.hdfs.minBlockReplicas=l al.sinks.kl.hdfs.fileType=DataStream al.sinks.kl.hdfs.writeFormat=Text al.sinks.kl.hdfs.rolllnterval=

17、60 al.sinks.kl.hdfs.rollSize=0 al.sinks.kl.hdfs.rollCount=0 al.sinks.kl,hdfs.idleTimeout=0 # Use a channel which buffers events in memory # 第 3 步: 酉己置数据通道 al.channels.cl.type = memory al.channels.cl.capacity = 1000al.channels.cl.transactionCapacity = 100 # Bind the source and sink to the channel # 第

18、 4 步： 将三者级联 al.sources.rl.channels = cl al.sinks.kl.channel = cl(3)启动 Flume Agentcd /usr/local/flume # nohup bin/flume-ng agent -n agentl -c conf -f conf/flume- conf.properties &通过上面介绍的一系列步骤，已经可以将Flume收集的数据输出到HDFSo3.整合 Flume 与 Kafka HDFS下面通过Sink设置的修改将Flume的日志输出到HDFS和Kafkao下面的IP地址只是示 例，用户根据具体需求改为生产环境

19、中的IP地址。#define sink begin#define the sink kl, 定义 HDFS 输出端 al.sinks.kl.type=hdfs al.sinks.kl.channel=cl al.sinks.kl.hdfs.useLocalTimeStamp=true al.sinks.kl.hdfs.path=hdfs:/192.168.11.174:9000/flume/events/%Y/%m/%d al.sinks.kl.hdfs.filePrefix=cmcc-%H al.sinks.kl.hdfs.fileType=DataStream al.sinks.kl.hd

20、fs.minBlockReplicas=lal.sinks.kl.hdfs.rolllnterval=3600 al.sinks.kl.hdfs.rollSize=0 al.sinks.kl.hdfs.rollCount=0al.sinks.kl.hdfs.idleTimeout=0 #define the sink k2,定义 Kafka 输出端al.sinks.k2.channel=c2 al.sinks.k2.type=com.cmcc.chiwei.Kafka.CmccKafkaSinkal.sinks.k2.metadata.broker.list=192.168.11.174:90

21、92A92.168.11.175:9092,192.168.11.176:9092 al.sinks.k2.partition.key=0al.sinks.k2.partitioner.class=com.cmcc.chiwei.Kafka.CmccPartitional.sinks.k2.serializer.class=Kafka.serializer.StringEncoderal.sinks.k2.request.required.acks=0 al.sinks.k2.cmcc.encoding=UTF-8al.sinks.k2.cmcc.topic.name=cmcc al.sink

22、s.k2.producer.type=asyncal.sinks.k2.batchSize=100#define sink end#以上配置将同样的数据无差异输出传递到多个输出端。al.sources.rl.selector.type=replicating本例配置了两个输出端：一个是输出到Kafka,为了提高性能，用内存通道。另一个是输 出到HDFS,离线分析。在配置文件中设置两个sink： 一个是Kafka的输出通道K2。一个是HDFS的输出通道K1。al.sources = rl al.sinks = kl k2 al.channels=cl c2#define channel begi

23、n#define the channel cl, al.channels.cl.type=fileal.channels.cl.checkpointDir=/home/flume/flumeCheckpointal.channels.cl.dataDirs=/home/flume/flumeData, /home/flume/flumeDataExt al.channels.cl.capacity=2000000 al.channels.cl.transactionCapacity=100 #define the channel c2 al.channels.c2.type=memory al

24、.channels.c2.capacity=2000000al.channels.c2.transactionCapacity=100 #define channel end#大家在配置文件中添加如上信息，即可配置好，同时输出到Kafka和HDFS。4.4.2用Kafka将日志汇总由于Flume收集的数据和后端处理的下游系统之间可能存在多对多的关系，为了解耦合 保证数据传输延迟，选用Kafka作为消息中间层进行日志中转。Apache Kafka是由Apache软件基金会开发的一个开源消息系统项目,由Scala写成。 Kafka最初是由Linkedln开发，并于2011年初开源。2012年10月

25、从Apache Incubator“毕业”。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台 lo它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范 的实现。Kafka进行消息保存时会根据Topic进行归类，发送消息者成为Producer,消息 接受者成为Consumer,此外Kafka集群有多个Kafka实例组成，每个实例(Server)成 为Brokero无论是Kafka集群，还是Producer和Consumer都依赖Zookeeper来保证系 统可用性集群保存一些元(Meta)信息，如图4-5所示。11.1 搜索引擎简介搜索排序概述11.2 查询

26、无关模型 PageRank基于 Spark 的分布式 PageRank 实现1141 PageRank 的 MapReduce 实现11.4.2 Spark 的分布式图模型 GraphX1143 基于 GraphX 的 PageRank 实现案例：GoogleWeb Graph 的 PageRank 计算11.3 杳询相关模型Ranking SVMSpark中支持向量机的实现11.3.1 Spark中的支持向量机模型使用Spark测试数据演示支持向量机的训练11.4 案例：基于MSLR数据集的查询排序Microsoft Learning to Rank 数据集介绍1182 基于 Spark 的

27、 Ranking SVM 实现本章小结图45 Kafka架构图1 .概念和术语消息：全称为Message,是指在生产者、服务端和消费者之间传输数据。 消息代理：全称为Message Broker,通俗来讲就是指该MQ的服务端或者服务器。 消息生产者：全称为Message Producer,负责产生消息并发送消息到meta服务器。 消息消费者：全称为Message Consumer,负责消息的消费。 消息的主题：全称为Message Topic,由用户定义并在Broker上配置。Producer发送消 息到某个Topic下，Consumer从某个Topic下消费消息。 主题的分区：也称为Part

28、ition,可以把一个Topic分为多个分区。每个分区是一个有序、 不可变的、顺序递增的Commit Log 消费者分组：全称为Consumer Group,由多个消费者组成，共同消费一个Topic下的消 息，每个消费者消费部分消息。这些消费者就组成一个分组，拥有同一个分组名称，通常 也称为消费者集群偏移量：全称为Offset。分区中的消息都有一个递增的id,称之为Offset。它唯一标识了 分区中的消息、。2 .部署 KafkaKafka安装中可以使用自带的Zookeeper,也可以使用外接的Zookeeper,本例以使用自 带的Zookeeper为例进行Kafka的部署和安装。(1)下载

29、KafkaKafka 官网下载安装包：http:/Kafka.apache.org/o解压：tar -zxvf Kafka_2.10-0.8.l.l.tgz(2)配置 Kafka 与 Zookeeper 文件1)酉己置 zookeeper.properties 文件。dataDir=/tmp/zookeeper clientPort=2181 maxClientCnxns=O initLimit=5 syncLimit=2 #以 下以三台为例，用户可以配置更多的服务器：server.43=10.190.172.43:2888:3888server.38=10,190.172.38:2888:3

30、888 server.33=10.190.172.33:2888:38882)酉己置 Zookeeper myido在每个服务器dataDir中创建myid文件，写入本机ID。例如：在server.43创建myid文件，并输入主机编号43。echo 43 /tmp/zookeeper/myid3)配置Kafka文件。配置config/server.properties,在每个节点根据不同的主机名进行 以下配置。broker.id： 43 #Unique； Write number, Config this nodes ID hostname：10.190.172.43#Unique,Serve

31、r IP, Config this node IP zookeeper.connect=10.190.172.43:2181,10.190.172.33:2181,10.190.172.38:2181(3) 启动 Zookeeper由于Kafka需要通过Zookeeper存储元数据信息，则预先启动Zookeeper,并提供给 Kafka相应连接地址。在每台服务器都执行命令：bin/zookeeper-server-start.sh config/zookeeper.properties(4)启动 Kafka启动命令：bin/Kafka-server-start.sh。(5)创建和查看Topic

32、为了测试部署的Kafka可用性，可以在Kafka中创建和查看Topic并进行可用性的验证。注意，这里的Topic和Flume中配置的要一致，同时后续Spark Streaming也是消费这个 Topic中的数据。bin/Kafka-topics.sh -create -zookeeper 10.190.172.43:2181 -replication-factor 1 - partitions 1 -topic KafkaTopic查看Topic,验证是否已经创建Topic,如果能够查到，则证明安装正确：bin/Kafka-topics.sh -describe -zookeeper 10.1

33、90.172.43:2181 3.整合 Kafka 与 Spark Streaming在用户的SBT项目中，在build.sbt中添加如下依赖。当应用进行项目编译，便会下载相 应驱动进行整合。libraryDependencies += Seq( org.scalatest % scalatest % 1.9.1 % test， org.apache.spark % spark-core % 1.2.0, org.apache.spark % spark-streaming % 1.2.0 norg.apache.sparkn % spark-streaming-Kafka1 %org.apa

34、che.Kafka %nKafkan%n0.8.1.1)1 Kafka 维基简介：http:/zh.wikipedia.org/wiki/Kafkao.3用Spark Streaming进行实时日志分析通过之前的整合，已经打通数据收集和中转的通道，数据通过Flume分别流向Kafka和 HDFS,在Kafka中的数据，由Spark Streaming消费并进行实时数据分析。对于进入 HDFS的数据，后续使用Spark SQL和MLlib进行离线数据分析。下面介绍Spark Streaming的实时日志分析。1.5 park Streaming 读取 Kafka 日志用户可以通过以下几个步骤和代

35、码示例构建读取日志的实时程序，然后启动Spark Streaming的程序并运行。代码示例通过Spark Streaming接收Kafka的输入数据，构建Kafka流式数据输入。进行输入参数的配置。初始化Spark Streaming的程序。进行日志分析，并启动Spark Streaming程序的运行。import org.apache.spark.streaming.Minutes, Seconds, StreamingContext import StreamingContext._ import java.util.Properties import org.apache.spark.s

36、treaming-import org.apache.spark.SparkConf def testKafkaWordargs: Array String)： Unit = val sparkConf =new SparkConfQ.setAppName,LogAnalysisu) val sc = new SparkContext(sparkConf) / 输入参数配置 Topic 和路径 val ArrayfzkQuorum, group, topics, numThreads, outPutHdfsPath, configFileprintOrWriteFile) = Array (a

37、rgs (0), args(l), args(2), args (3), args(4; args(5), args(6) / Spark Streaming 初始化 val ssc = new StreamingContextfsc,Seconds(2)J ssc.checkpointHcheckpointn val topicMap =topics.splitJ),map(-numThreads.toInt),toMap val logs = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)进行日志的深入分析与处理，并启动Spa

38、rkStreaming 的程序进行运行。doSomeProcessQ ssc.startQ ssc.awaitTerminationQ 2 .实时点击流分析下面给出一些示例进行点击流(ClickStream)分析，用户可以根据相应示例进行复杂示 例拓展。(1)预定义定义PageView类，记录站点ID、访客ID、网页的URL。class PView(val site:String,val visitor:String, val pageurl:String) extends Serializable 将每条数据格式化后转换为PageViews对象，便于后续的分析。val pageviews =

39、 logs.map(PView.parseData(_)(2)统计分析1）统计每批次指定时间段数据PV。时间长度根据 Streamingcontext (sc, Seconds (TimeDuration)中的 TimeDuartion 进 行设置。val pagecounts = pageviews.map(view = view.pageurlj.countByValueQ2)统计过去15s的访客数量，每隔2s计算一次。val window = Seconds(15) val interval = Seconds(2) val visitorcounts = pageviews.windo

40、wwindow,interval.map(view = (view.visitor, l),groupByKey().map(v = (v._l,v._2.size.实时百分位(Percentile)统计每个用户中段访问量的页面网站的日常日志分析中，常用的日志分析场景需要观察页面的加载时间：如果页面加载时 间过长可能是因为网络的问题，或者服务器的问题；如果页面时间很短则不需要进行优化。 往往需要统计加载时间处于中游的页面并进行分析，这样对指定页面进行JS或者后台的 优化。这个分析应用的思路是，统计每个页面和浏览器组的加载延迟，最后统计每个页面和浏览 器的加载延迟中排在25%, 50%, 75%

41、的时间是多长。之后再根据这些信息进行页面的 优化和更深入的统计。import org.apache.spark.storage.StorageLevel importorg.apache.spark.streaming.StreamingContext, Seconds import org.apache.spark.streaming.Kafka.KafkaUtils import org.apache.spark.SparkContext, SparkConf import Kafka.serializer.Decoder, StringDecoder importscala.collec

42、tion.mutable import scala.collection.mutable.HashMap def Percentilefargs:Array String): Unit=将数据预处理为包含浏览器、页面URL和加载延迟的三元组val filteredLog = logs.map(mapFunction) val result = filteredLog.maprecord = val s = record._2.splitn) / s(0)代表浏览器类型,s(l)代表页面 URL val key = Seq(s(0), s)/ s(l)代表加载时间 val value = s(2

43、.tolnt key, value) ) ,mapPartitionsfiter = / Map Side 聚集汇总结果 HashMap is Map (key - Map(value, count) val resultMap = new HashMap Seq String, HashMap Int, I nt var tmp:(Seq String, I nt) = null while(iter.hasNext) tmp = iter.nextQ val valueMap = resultMap.getOrElsetmp._l, new HashMapInt, I nt var cou

44、nt = valueMap.get0rElse(tmp._2, 0) valueMap.put(tmp._2, count + 1) resultMap.putmp.l, valueMap) resultMap.iterator .reduceByKeyAndWindow(x: HashMapInt, Int, y: HashMapInt, Int) = / Reduce 端聚集汇总结果，合并两个 Hash 表。y.foreachr = x.put(r._l, x.getOrElse(r._l, 0) + r._2) ) x , timeDuration, timeDuration).mapP

45、artitionsiter = / 计算百分位，结果格式为：Map (key, Map(percentage/ value) val resultMap = new mutable.HashMapSeqString, mutable.HashMapDouble, IntQ while(iter.hasNext) val tmp = iter.next val map = tmp._2 val sumCount = map.map(r = r._2.reduce(_+J val p25 = sumCount * 0.25 val p50 = sumCount * 0.5 val p70 = su

46、mCount * 0.75 val sortDataSeq = map.toSeq.sortByfr = r._l) val iterS = sortDataSeq.iterator var curTmpSum = 0.0 var prevTmpSum = 0.0 val valueMap =new mutable.HashMapDouble, IntQ while(iterS.hasNext) val tmpData = iterS.nextQ prevTmpSum = curTmpSum curTmpSum += tmpData._2 iffprevTmpSum = p25 valueMap.put(0.25, tmpData._l else ifprevTmpSum = p50) valueMap.put(0.5, tmpData._l) else iffprevTmpSum = p50 valueMap.put(0.75, tmpData._l resultMap.puttmp._l, valueMap resultMap.iterator ) / 结果输出 if(printOrWriteFile.toBoolean) result.print else resu