# RealTime

**Repository Path**: hanhan-jun/real-time

## Basic Information

- **Project Name**: RealTime
- **Description**: 实时计算的电商大数据处理
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2022-01-08
- **Last Updated**: 2022-01-08

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# RealTime

#### 介绍
实时计算的电商大数据处理

#### 项目背景

某在线电子图书商城当前注册会员数有近千万会员，遍布于全国多个省份，日常在线业务量较大。为了提高商城的用户知名度，该公司拟计划开展大型促销活动（类似于双十一的电商购物活动）。

为了准确、实时统计当前销售数量，公司拟委托你开发一个促销活动的大数据看板，用于当日活动启动后销售订单实时数据的变化情况。

为提供更真实的测试环境，公司的市场部门委托相关人员已设计了一个订单数据模拟器，可模拟产生客户在网上下单的订单信息，数据会自动存入 MySQL数据库相应的表中。

该模拟器允许调节进程的数量，模拟不同量级的并发量，以充分测试系统的性能。一共有2张表，第一张表是描述订单中所购图书的详细信息，第二张表是订单总体信息，包括配送信息。

![第一张表](image.png)
![第二张表](image1.png)



#### 总体需求
本次大作业也是一个模拟电商平台销售书籍数据实时分析。那么这就要求计算的时延必须小。有时候数据量大、计算复杂的原因会导致实时计算无法完成，这时甚至会牺牲结果的准确性，在保证误差在可接受范围的前提下，优先满足计算的实时性。而笔者也会在搭建的框架上去尽可能满足这个需求！

拟完成功能的需求如下：

运用实时计算技术、Web 技术构建一个图书商城的促销活动大数据看板，并实现以下功能：

（1）界面每 3 秒刷新一次，如果能做到 2 秒或 1 秒刷新一次

（2）展示当前订单的已处理速度，单位为“条/秒”

（3）大数据看板界面应简洁大气，有特点

（4）展示截止当前时间的总销售金额、订单数量、下单客户数基本信息

（5）展示截止当前时间销量排名前10的图书排行榜

（6）展示截止当前时间销量排名前10的出版社排行榜

（7）展示全国各地下单客户的累计数量（按省份），在地图上直观展示

（8）数据统计误差（数据丢失、统计错误）不超过1%，应设计实验计算数据误差率

（9）展示的数据延迟应不超过30秒，每次刷新时应显示获取的数据最新时间

（11）测试出系统的最大承载负荷量，即最大并发线程数

（12）如果实现了以上规定之外的有特色数据展示功能



#### 方案实现

在我看来，方案的选择就是各种组件的组合使用，再考虑性能。一个是数据产生组件，一个消息中间件，一个实时流处理组件，一个存储仓库，一个数据可视化组件。数据产生组件可是storm的spout或则flume亦或者canal亦或者是kafka，消息中间件就是kafka，实时流计算框架是Spark Streaming、storm、flink，而存储仓库可以用MongoDB、Redis、Mysql，数据可视化可以用flask、echarts。所以方案选取有很多，相互组合即可

第一种方案：

flume+kafka+storm+redis+flask。首先在flume中的创建两个.conf文件（因为两个表，两个topic）分别定义source，channal、sink三个组件，在配置source时，功能是从本地mysql读取数据，由于source没有sql，需要导入一个sqlsource的jar包，所以类型是sqlsource。在配置sink时，类型为kafka，并创建相应话题，之后kafka话作为消息中间件在flume和storm之间传输数据。在storm项目中定义一个spout消费kafka中的数据并切词，之后在bolt中根据字段数据进行处理，把处理得到的数据储存到redis中，然后利用flask把redis相应数据发送大数据看板。


第二种方案：

canal+kafka+storm+redis+flask。这一种方案较第一种方案，将flume换成了canal，canal直接读取binlog文件，解析数据，通过配置借用消息队列kafka消息订阅传到storm，其余操作与第一种一样。

最终，我选择了第二种方案，第一种方案的可行性不强，flume查询mysql数据，做不到从最新的数据开始读取，并且是循环读取，虽然可以通过要查询的表的自增变量来实现，但是不适用于本次项目。没办法，发现这条路行不通过后，我选择了canal读取数据，之后要是还有相关的项目，可以再试试flume。


#### 方案流程图

![流程图](image2.png)


#### 单元实现

1.数据采集：

在数据采集单元中，我们主要通过老师发数据生成器read.exe产生数据，先按照说明，在本地安装mysql，我安装的是8.0.27版本的mysql，然后再安装了一个可视化工具NaviCat，之后下载canal的安装包，在本地或者虚拟机安装都行，我是先在虚拟机安装的canal，后面canal跑着跑着就出现问题，无法解决，卸载安装到了windows本地，配置好bind的ip地址，创建mysql的canal角色和密码，启动canal，读取mysql的binlog文件。

2.数据订阅：

在本单元中，我们通过搭建kafka集群来实现数据订阅，集群的安装也十分简单，下载安装包，我下载的是对应Scala2.11版本的kafka_2.11-2.4.1，建立zk集群，在这我是配置了三台虚拟机搭建集群，之后便是各种文件的配置，总的来说配置还是十分简单方便的，一切准备就绪之后开始kafka服务进行部署，在数据采集单元的canal配置中，我们创建了两个订阅，即两个topic，相应的，我们需要在kafka集群中建立两个topic，kafka的消息订阅功能真的厉害，这个机制保障了我们的数据分发。现在已经有了topic和producer，我们需要storm来消费数据。一开始我是按照网上的storm消费kafka教程，调用低版本kafka中封装好的KafkaSpout接口，输入spoutconfig参数，但是刚开始并不很清楚其中的流程，按照参数输入zk集群的ip地址以及topic和分组id，帮不能正常消费数据，现在回想起来，原因出在spout的emit值是什么类型，在书写bolt代码的时候没法从spout中接受数据。最后我是在spout中实现consumer代码，如图所示。

 ![图 4 consumer实现](image3.png)

连接成功之后，建立ConsumerRecords对象，并且创建一个ConsumerRecords类型的迭代器iterator进行迭代获取数据，只要迭代器下一个不为空就进行数据处理。

3.数据处理：

本单元的实现是在数据订阅中提到过的spout中，获取迭代器中的数据，将其转换成字符串类型，之前提到过，对order表读入的数据利用jsonObject方法将字符串数据转JSON数据，获取data字段的数据之后再通过json格式的键值来切分数据，之后便是通过emit方法传送value，如图所示。

 ![图 5 切分order表数据](image4.png)

同时，实现接口方法declareOutputFields，需要声明传送值value的字段，如图所示。

 ![图 6 实现declareOutputFields方法](image5.png)

同样的，还需要创建一个订阅orderdetail表的spout，大致过程与order一致，但是切分方式不同，如图所示。

 ![图 7切分orderdetail表](image6.png)

Spout已经创建好，接下来就是在bolt中按字段接受后存入redis数据库中。首先我们要搭建好redis的环境，在windows本地或则虚拟机中都行，重点是配置文件中的bind：IP地址，以及不用设置requirepass即密码。还需引入相应依赖，再创建一个JeidsUtils工具类，里面有redis的各种方法函数。一切准备就绪，连接本机redis，bolt按字段获取到数据后，通过zincrby或incryby等方法存入redis，主要获取数据是书名，出版社，下单数量，书的类型，单价，收货人和收货地址，订单总金额。如图1和图2所示。

 ![图 8.1 order数据存入redis](image7.png)

                                       图 1 order数据存入redis

 ![图表 8.2 orderdetail数据存入redis](image8.png)

                                       图 2 orderdetail数据存入redis


4.数据展示：

通过数据处理，数据都存储在了redis中，我们只需通过redis的一些方法实现增量查询后，将数据输出到web端，主要实现还是python连接redis，以及flask传送数据到Ajax，Ajax将数据输送到web界面展示。主要问题还是数据的展示形式，也就是web端的界面设计。

5.性能调优：

性能调优主要依靠并行度的调整，我们通过topology中的setSpout和setBolt两个方法设置并行度，如图所示。

 ![图 9 Topology调整并行度](image9.png)

                                      图 Topology调整并行度

通过storm的ui界面查看性能指标，如图所示。

 ![图 10 性能指标](image10.png)

①	Execute latency：消息的平均处理时间，单位是毫秒。

②	Process latency：消息从收到到被ack掉所花费的时间，单位为毫秒。如果没有启用Acker机制，那么Process latency的值为0。

③	Capacity：计算公式为Capacity = Bolt 或者 Executor 调用 execute 方法处理的消息数量 × 消息平均执行时间/时间区间。如果这个值越接近1，说明Bolt或者 Executor 基本一直在调用 execute 方法，因此并行度不够，需要扩展这个组件的 Executor数量。

Execute latency,Process latency是处理消息的时效性，而Capacity则表示处理能力是否已经饱和。从这3个参数可以知道Topology的瓶颈所在。