环境部署

相关环境部署的笔记如下：

1. zookeeper和spark安装：在zookeeper上搭建Spark集群的小笔记

2. kafka安装和使用：kafka学习笔记

3. sbt下载和安装：Linux无坑安装sbt

项目主要架构

数据总体上可以分为静态数据和流数据。对静态数据和流数据的处理，对应着两种截然不同的计算模式：批量计算和实时计算。批量计算以“静态数据”为对象，可以在很充裕的时间内对海量数据进行批量处理，计算得到有价值的信息。Hadoop就是典型的批处理模型，由HDFS和Hbase存放大量的静态数据，由MapReduce负责对海量数据执行批量计算。流数据必须采用实时计算，实时计算最重要的一个需求是能够实时得到计算结果，一般要求响应时间为秒级。当只需要处理少量数据时，实时计算并不是问题；但是，在大数据时代，不仅数据格式复杂、来源众多，而且数据量巨大，这就对实时计算提出了很大的挑战。因此，针对流数据的实时计算——流计算，应运而生。

流计算处理过程包括数据实时采集、数据实时计算和实时查询服务。

• 数据实时采集：数据实时采集阶段通常采集多个数据源的海量数据，需要保证实时性、低延迟与稳定可靠。目前有许多互联网公司发布的开源分布式日志采集系统均可满足每秒数百MB的数据采集和传输需求，如Kafka和Flume等。

• 数据实时计算：流计算处理系统接收数据采集系统不断发来的实时数据，实时地进行分析计算，并反馈实时结果。
• 实时查询服务：流计算的第三个阶段是实时查询服务，经由流计算框架得出的结果可供用户进行实时查询、展示或储存。（一般是把计算结果实时地推送给用户）

1. 实时计算架构

   • python读取csv文件，传送数据到kafka
   • spark-streaming和kafka集成
   • spark-streaming读取传送到kafka的数据，进行实时统计，统计后的数据再推送到kafka
   • flask实时接收处理好的kafka数据，利用flask-socketio推送到客户端，进行绘图展示

2. 批量计算架构

   • 把待分析的数据集导入hive
   • hive编写HQL语句批量分析数据
   • sqoop把hive中分析后的结果导入mysql中
   • flask读取mysql数据，利用echarts绘图进行可视化分析

具体步骤

流计算步骤

在统计淘宝双11的各省份的实时销量、淘宝双11的实时年龄分布、淘宝双11的各购物行为所占的比重、淘宝双11那天的日活量均采用流计算。

python连接kafka

在这一步中，python读取csv文件并连接kafka，把数据传送过去

需要安装pykafka、pandas等第三方库

python连接kafka代码

执行上述代码之后可在kafka集群这边任意一个节点上测试是否能接收到数据，从而判断数据是否成功传送到kafka。（以实时省份销量数据为例）

spark-streaming集成kafka

Spark Streaming的基本原理是将实时输入数据流以时间片（秒级）为单位进行拆分，然后经Spark引擎以类似批处理的方式处理每个时间片数据，能做到秒级响应，不能做到毫秒级响应。Spark Streaming实际上是仿照流计算，并不是真正实时的流计算框架。

kafka安装：kafka学习笔记，为和下面的spark-streaming-kafka集成jar包适配，建议装0.8或者0.10版本的kafka（分情况，看spark版本）。

下载spark-streaming-kafka的jar包：

spark2.3.0版本以下（不包括2.3.0）的可以下载spark-streaming-kafka-0-8的集成：spark-streaming-kafka-0-8，选择与自己scala和spark版本相对应的jar包下载

spark2.3.0以上版本（包括2.3.0）的可以下载spark-streaming-kafka-0-10以上的集成：spark-streaming-kafka-0-10

把下载的jar包拷贝到/usr/local/spark/jars下

在/usr/local/spark/jars目录下，新建一个kafka目录，把kafka安装目录的libs目录下的所有jar文件复制到/usr/local/spark/jars/kafka目录下，

下面测试spark和kafka环境是否连通

启动spark-shell

在spark-shell里引入如下jar包

若引入jar包不出错，说明kafka和spark环境已经连通

编写并运行spark-streaming程序(实时词频统计)

scala的官方API文档：https://www.scala-lang.org/api/2.11.12/#scala.package，此文档对应的scala版本是2.11.12，也可以到：https://docs.scala-lang.org/api/all.html 查找与自己的scala版本相对应的API文档。

spark的官方API文档：http://spark.apache.org/docs/2.4.5/api/scala/index.html#org.apache.spark.package，此文档对应的scala版本是2.11.12，也可以到：http://spark.apache.org/docs/ 查找与自己的spark版本相对应的API文档。

创建项目spark_connect_kafka

编写scala代码

其中有一句"reduceByKeyAndWindow(+,-, Seconds(1), Seconds(1), 1)"，关于reduceByKeyAndWindow算子所执行的操作可以戳：SparkStreaming滑动计算窗口reduceByKeyAndWindow图解说明，说通俗点，可以把每一次窗口以内的数据看作是一个池塘里的鱼，当窗口以指定的滑动时间滑动时，每一次滑动都有鱼游入和游出，当有鱼游入时，就加上游入的该类鱼的数量，当有鱼游出去时，就减去游出的该类鱼的数量，以此统计出每一次窗口内的每类鱼的数量。

这样做是为了减少计算量，通过只计算游出去多少鱼和游进来多少鱼而得到当前池塘（窗口）的鱼的数量，而不必把整个池塘（窗口）都统计一遍，减少了窗口长度-（2*滑动时间）的运算量。

编写sbt文件

然后编译并打包程序

打包成功之后，可以编写运行脚本，在/usr/local/spark/mycode/spark_connect_kafka目录下新建startup.sh文件，输入如下内容（以统计各省份实时销量时的参数为例，在统计其他类型的数据时，需自行更改spark-streaming消费的topics和生产的topics）

最后在/usr/local/spark/mycode/spark_connect_kafka目录下，运行如下命令即可执行刚编写好的Spark Streaming程序

编写并运行spark-streaming程序(累加词频统计)

spark-streaming的累加词频统计其他均与spark-streaming的实时词频统计一致，只是scala代码中增加了updateStateByKey算子来累加上一批次计算的结果。

在kafka上查看数据统计结果

开启消费，看是否能消费到上面spark-streaming程序生产出的topics消息（以省份统计数据为例）

批量计算步骤

用于统计淘宝双11销量前20的商品类型ID

往hive中导入数据

启动hive

根据数据集字段创建相应的表

导入数据

筛选出销量前20的商品类型ID

sqoop导出数据到mysql

在mysql上创建相应的数据库和数据表

sqoop导出数据到mysql

mysql这边查看，可以看到数据成功导入

淘宝RFM用户划分

选取11月份之前的用户购买记录数据进行RFM用户划分，为每一个用户打上标签。从而统计双11活动周的营销策略在唤回用户方面的效果。

在这里由于数据集不包含用户消费金额，所以用户划分只从R、F两个维度来进行划分。

只选取11月份之前的用户购买记录来进行RFM用户模型建模是因为：双11是特殊的一天，几乎所有的用户在双11那天都有消费记录，而数据集的数据记录又只记录到11月12日那天，如果选取包含11月份的数据在内的数据进行RFM用户模型建模，那么几乎所有的用户的R值都会趋于一致，用户在R维度上的区别都会被抹平。

Flask-SocketIO实时推送数据

在流计算步骤和批量计算步骤获得所需的数据后，可利用flask读取kafka和mysql中的数据，利用flask-socketio把数据实时推送到客户端，客户端通过socket.io.js实时接收数据，利用echarts进行动态绘图展示

flask应用程序代码，一个主线程三个子线程

Echarts动态绘图

项目结果