文章作者邮箱：yugongshiye@sina.cn              地址：广东惠州

 ▲ 本章节目的

⚪ 了解TELECOM项目的数据收集；

⚪ 了解TELECOM项目的数据清洗；

⚪ 了解TELECOM项目的数据导出；

⚪ 了解TELECOM项目的数据可视化；

⚪ 了解TELECOM项目的其他；

一、数据收集

1. 在实际生产环境中，电信流量日志肯定不只是在一台服务器上产生，而是每一台服务器都会产生流量日志。所以此时，需要先搭建Flume的扇入流动模型，之后将收集到的数据传输到HDFS上进行存储。

2. 步骤：

a. 在第二台和第三台服务器上来创建对应的目录用于存储日志(将第二台和第三台服务器当作是日志产生的服务器) 。

cd /home

mkdir telecomlog

b. 进入对应的目录，将日志上传或者下载到指定目录下(实际过程中，日志肯定是实时产生的) 。

cd telecomlog/

#云主机的下载地址

 wget http://bj-yzjd.ufile.cn-north-02.ucloud.cn/103_20150615143630_00_00_000_2.csv

c. 收集第二台和第三台服务器上的日志，将收集到的日志传输到第一台服务器上，进行数据的扇入。

cd /home/software/apache-flume-1.9.0-bin/data

#编辑文件

vim telecomlog.conf

#在文件中添加如下内容

a1.sources = s1

a1.channels = c1

a1.sinks = k1

# 日志是放在指定目录下的

# 所以此时监控指定目录下的变化

# 如果目录下产生了新的文件

# 需要收集这个新文件中的内容

a1.sources.s1.type = spooldir

# 指定要监听的目录

a1.sources.s1.spoolDir = /home/telecomlog

# 配置Channel

a1.channels.c1.type = memory

a1.channels.c1.capacity = 10000

a1.channels.c1.transactionCapacity = 1000

# 需要将收集到的数据发送到第一台服务器上

a1.sinks.k1.type = avro

a1.sinks.k1.hostname = hadoop01

a1.sinks.k1.port = 8090

# 绑定

a1.sources.s1.channels = c1

a1.sinks.k1.channel = c1

d. 数据收集到第一台服务器上之后，需要将收集到的数据写到HDFS上。

cd /home/software/apache-flume-1.9.0-bin/data/

#编辑文件

vim telecomlog.conf

#在文件中添加如下内容

a1.sources = s1

a1.channels = c1

a1.sinks = k1

# 需要接收第二台和第三台服务器传输来的数据

a1.sources.s1.type = avro

a1.sources.s1.bind = 0.0.0.0

a1.sources.s1.port = 8090

# 需要在数据中来添加一个时间戳

a1.sources.s1.interceptors = i1

a1.sources.s1.interceptors.i1.type = timestamp

# 配置Channel

a1.channels.c1.type = memory

a1.channels.c1.capacity = 10000

a1.channels.c1.transactionCapacity = 1000

# 配置Sink

# 需要将数据写到HDFS上，最好能够实现数据按天存储

a1.sinks.k1.type = hdfs

# 指定数据在HDFS上的存储路径

a1.sinks.k1.hdfs.path = hdfs://hadoop01:9000/telecomlog/reporttime=%Y-%m-%d

# 指定文件在HDFS上的存储类型

a1.sinks.k1.hdfs.fileType = DataStream

# 指定文件的滚动时间间隔

a1.sinks.k1.hdfs.rollInterval = 3600

a1.sinks.k1.hdfs.rollSize = 0

a1.sinks.k1.hdfs.rollCount = 0

# 绑定

a1.sources.s1.channels = c1

a1.sinks.k1.channel = c1

e. 启动HDFS。

start-dfs.sh

f. 启动第一台服务器上的Flume。

../bin/flume-ng agent -n a1 -c ../conf -f telecomlog.conf -

Dflume.root.logger=INFO,console

g. 启动第二台服务器和第三台服务器上的Flume。

../bin/flume-ng agent -n a1 -c ../conf -f telecomlog.conf -

Dflume.root.logger=INFO,console

二、数据清洗

1. 利用Flume将数据收集到了HDFS上，那么此时需要在Hive中建表来管理原始数据。

#启动YARN

start-yarn.sh

#进入HBase的安装目录的lib目录下

cd /home/software/hbase-2.4.2/lib

#进入子目录

cd client-facing-thirdparty/

#重命名

mv commons-logging-1.2.jar commons-logging-1.2.bak

mv log4j-1.2.17.jar log4j-1.2.17.bak

mv slf4j-log4j12-1.7.30.jar slf4j-log4j12-1.7.30.bak

#启动Hive的服务进程

hive --service metastore &

hive --service hiveserver2 &

#进入hive的客户端

hive

#创建库

create database telecom;

#使用这个库

use telecom;

#建表来管理原始数据

create EXTERNAL table telecom (a1 string, a2 string, a3 string, a4 string, a5 string, a6 string, a7 string, a8 string, a9 string, a10 string, a11 string, a12 string, a13 string, a14 string, a15 string, a16 string, a17 string, a18 string, a19 string, a20 string, a21 string, a22 string, a23 string, a24 string, a25 string, a26 string, a27 string, a28 string, a29 string, a30 string, a31 string, a32 string, a33 string, a34 string, a35 string, a36 string, a37 string, a38 string, a39 string, a40 string, a41 string, a42 string, a43 string, a44 string, a45 string, a46 string, a47 string, a48 string, a49 string, a50 string, a51 string, a52 string, a53 string, a54 string, a55 string, a56 string, a57 string, a58 string, a59 string, a60 string, a61 string, a62 string, a63 string, a64 string, a65 string, a66 string, a67 string, a68 string, a69 string, a70 string, a71 string, a72 string, a73 string, a74 string, a75 string, a76 string, a77 string) partitioned by (reporttime string) row format delimited fields terminated by '|' stored as textfile location '/telecomlog';

#修复分区

msck repair table telecom;

#数据抽样

select * from telecom tablesample(5 rows);

2. 原始数据中包含了77个字段，加上分区在内，一共有78个字段，但是这78个字段不代表都是需求中所需要的，所以此时需要从这77个字段中来抽取真正有用的字段来进行下一步的处理 - 此时需要对数据进行第一次的简单的清洗。

序号	字段名	类型	长度	说明
16	TAC	byte	2	TAC
17	Cell ID	Byte	4	UE所在小区的ECI
19	App Type Code	byte	1	业务类型编码，参见附录D XDR类型编码定义
20	Procedure Start Time	dateTime	8	TCP/UDP流开始时间，UTC时间），从1970/1/1 00:00:00开始到当前的毫秒数。
21	Procedure End Time	dateTime	8	TCP/UDP流结束时间，UTC时间），从1970/1/1 00:00:00开始到当前的毫秒数
23	App Type	byte	2	应用大类，集团规定的18种应用大类，参见《XXXX数据流量DPI识别能力规范》
24	App Sub-type	byte	2	应用小类，根据集团定义的识别规则识别出来的小类, 参见《XXXX数据流量DPI识别能力规范》。集团未定义的各厂家根据自己的DPI进行识别
27	USER_IPv4	byte	4	终端用户的IPv4地址，如无则填全F
29	User Port	byte	2	用户的四层端口号
31	App Server IP_IPv4	byte	4	访问服务器的IPv4地址，如无则填全F
33	App Server Port	byte	2	访问的服务器的端口
34	UL Data	byte	4	上行流量 单位：字节 对于场景一，定义为从内层IP包头开始计算的数据包大小总和；对于其他场景，定义为从链路层封装开始计算的数据包大小总和
35	DL Data	byte	4	下行流量 单位：字节 对于场景一，定义为从内层IP包头开始计算的数据包大小总和；对于其他场景，定义为从链路层封装开始计算的数据包大小总和
40	上行TCP重传报文数	byte	4	上行TCP重传报文数，非TCP传输时，此字段填0
41	下行TCP重传报文数	byte	4	下行TCP重传报文数，非TCP传输时，此字段填0
55	HTTP/WAP事务状态	byte	2	HTTP/WAP2.0层的响应码，参见附录A 状态编码
59	HOST	char	64	访问域名
62	User-Agent	char	256	终端向访问网站提供的终端信息，包括IMEI、浏览器类型等
63	HTTP_content_type	char	128	HTTP的内容是文字还是图片、视频、应用等，具体编码参考附录A
68	Wtp中断类型	byte	1	WTP层的失败类型
72	业务行为标识	byte	1	0-业务登陆 1-页面访问 2-刷新 3-未识别 判断规则详见《业务KPI定义(20130821)》
73	业务完成标识	byte	1	0-业务成功 1-业务失败 2-未识别 成功的判断规则：状态码<400

确定好要抽取的字段之后，需要建立一个表来存储这些被抽取的字段。

#建立抽取的字段表 - 事实表

create table dataclear(reporttime string, appType bigint, appSubtype bigint, userIp string, userPort bigint, appServerIP string, appServerPort bigint, host string, cellid string, appTypeCode bigint, interruptType String, transStatus bigint, trafficUL bigint, trafficDL bigint, retranUL bigint, retranDL bigint, procdureStartTime bigint, procdureEndTime bigint)row format delimited fields terminated by '|';

#从原始表中来抽取字段放入事实表中

insert overwrite table dataclear select concat(reporttime, ' ', '00:00:00'), a23, a24, a27, a29, a31, a33, a59, a17, a19, a68, a55, a34, a35, a40, a41, a20, a21 from telecom;

#抽样数据

select * from dataclear tablesample(5 rows);

3. 抽取完字段之后，需要对字段整理，例如格式的转化，字段的整合，字段的填充等。

序号	字段	字段类型	描述
0	reportTime	datetime	小时 时间片 default 'YYYY-MM-DD HH24:MI:SS'
1	appType	int	应用大类
2	appSubtype	int	应用小类
3	userIP	varchar(20)	用户IP
4	userPort	int	用户端口
5	appServerIP	varchar(20)	服务器IP
6	appServerPort	int	服务器端口
7	host	varchar(50)	域名
8	cellid	varchar(20)	小区ID
9	attempts	int(20)	尝试次数
10	accepts	int(20)	接受次数
11	trafficUL	int(20)	上行流量
12	trafficDL	int(20)	下行流量
13	retranUL	int(20)	重传上行报文数
14	retranDL	int(20)	重传下行报文数
15	failCount	int(20)	延时失败次数
16	transDelay	int(20)	传输时延*

建表来存储整理好格式之后的数据。

cd sqoop-1.4.7/conf/

e. 复制文件。

 

cp sqoop-env-template.sh sqoop-env.sh

#编辑文件

vim sqoop-env.sh

#在文件尾部添加

export HADOOP_COMMON_HOME=/home/software/hadoop-3.1.3

export HADOOP_MAPRED_HOME=/home/software/hadoop-3.1.3

export HIVE_HOME=/home/software/hive-3.1.2

export ZOOKEEPER_HOME=/home/software/apache-zookeeper-3.5.7-bin

export ZOOCFGDIR=/home/software/apache-zookeeper-3.5.7-bin/conf

#保存退出，重新生效

source sqoop-env.sh

f. 进入Sqoop安装目录的lib目录。

cd ../lib

g. 添加MySQL的连接驱动包。

 wget http://bj-yzjd.ufile.cn-north-02.ucloud.cn/mysql-connector-java-5.1.27.jar

h. 配置环境变量。

vim /etc/profile

#在文件末尾添加

export SQOOP_HOME=/home/software/sqoop-1.4.7

export PATH=$PATH:$SQOOP_HOME/bin

#保存退出，重新生效

source /etc/profile

#检查是否配置正确

sqoop version

3. Sqoop的基本命令：

命令	解释
sqoop version	查看Sqoop的版本
sqoop list-databases --connect jdbc:mysql://hadoop01:3306/ -username root -password root	查看MySQL中的所有的database
sqoop list-tables --connect jdbc:mysql://hadoop01:3306/hive --username root --password root	查看MySQL中hive库下的所有的表
在MySQL中建表 create table person(id int primary key, name varchar(20), age int, gender varchar(20)); 在表中插入数据 导出到HDFS上 sqoop import --connect jdbc:mysql://hadoop01:3306/test --username root --password root --table person --target-dir '/sqoop/person' --fields-terminated by '\t' -m 1;	将MySQL中的数据导入到HDFS上
在MySQL中建表 create table orders(orderid int primary key, orderdate varchar(10), productid int, num int); 导出到MySQL中 sqoop export --connect jdbc:mysql://hadoop01:3306/test --username root --password root --export-dir '/orders/order.txt' --table orders -m 1 --fields-terminated-by ' ';	将HDFS上的数据导出到MySQL中

2. 导出数据

1. 在MySQL来创建的对应的库和表。

create database telecom;

use telecom;

2. 在MySQL建表。

#建立总表

create table F_HTTP_APP_HOST( reporttime datetime, appType bigint, appSubtype bigint, userIP varchar(20), userPort bigint, appServerIP varchar(20), appServerPort bigint, host varchar(255), cellid varchar(20), attempts bigint, accepts bigint, trafficDL bigint, trafficUL bigint, retranUL bigint, retranDL bigint, failCount bigint, transDelay bigint) ;

#建立各个APP的受欢迎程度表

create table D_H_HTTP_APPTYPE(hourid datetime, appType int, appSubtype int, attempts bigint, accepts bigint, succRatio double, trafficUL bigint, trafficDL bigint, totalTraffic bigint, retranUL bigint,retranDL bigint, retranTraffic bigint, failCount bigint, transDelay bigint) ;

3. 将HDFS上的数据导出到MySQL中

sqoop export --connect jdbc:mysql://hadoop01:3306/telecom --username root --password root --export-dir '/user/hive/warehouse/telecom.db/f_http_app_host/000000_3' --table F_HTTP_APP_HOST -m 1 --fields-terminated-by '|'

sqoop export --connect jdbc:mysql://hadoop01:3306/telecom --username root --password root --export-dir '/user/hive/warehouse/telecom.db/d_h_http_apptype/000000_0' --table D_H_HTTP_APPTYPE -m 1 --fields-terminated-by '|'

四、数据可视化

1. 概述

1. 数据导出到MySQL之后，需要进行可视化，可视化的目的是为了让数据进行更直观的展现，例如可以以折线图或者曲线图的形式来展现数据的增长趋势，可以用圆形图或者饼形图的形式来展现各部分数据所占的比例。

2. 市面上可视化的工具非常多，例如说Echarts、HighCharts等工具都可以进行可视化，同时市面上大部分的BI工具也都提供了可视化功能，例如TableAU，Power BI等机制。

2. 步骤

1. 需要在Windows系统中来安装MySQL的连接驱动。

2. 需要在Windows系统中来安装Power BI工具。

3. 配置连接。

4. 配置凭据。

5. 建立层次结构，然后进行比较。

五、其他

1. 各个网站的表现能力

1. 字段

序号	字段	字段类型	描述
0	hourid	datetime	小时时间片
1	host	varchar(50)	域名
2	appServerIP	varchar(20)	服务器IP
3	attempts	int(20)	尝试次数
4	accepts	int(20)	接受次数
5	succRatio	int(20)	尝试成功率
6	trafficUL	int(20)	上行流量
7	trafficDL	int(20)	下行流量
8	totalTraffic	int(20)	总流量
9	retranUL	int(20)	重传上行报文数
10	retranDL	int(20)	重传下行报文数
11	retranTraffic	int(20)	重传报文数据
12	failCount	int(20)	延时失败次数
13	transDelay	int(20)	传输时延

2. 建表语句

create table D_H_HTTP_HOST(hourid datetime, host varchar(255), appserverip varchar(20), attempts bigint, accepts bigint, succratio bigint, trafficul bigint, trafficdl bigint, totaltraffic bigint, retranul bigint, retrandl bigint, retrantraffic bigint, failcount bigint, transdelay bigint);

3. 插入语句

insert into D_H_HTTP_HOST select reporttime as hourid, host, appserverip, sum(attempts) as attempts, sum(accepts) as accepts, sum(accepts)/sum(attempts) as succRatio, sum(trafficul) as trafficul, sum(trafficdl) as trafficdl, sum(trafficul)+sum(trafficdl) as totaltraffic, sum(retranul) as retranul, sum(retrandl) as retrandl, sum(retranul)+sum(retrandl) as retrantraffic, sum(failcount) as failcount, sum(transdelay) as transdelay from F_HTTP_APP_HOST group by reporttime, host, appserverip;

2. 小区HTTP上网能力

1. 字段

序号	字段	字段类型	描述
0	hourid	datetime	小时时间片
1	cellid	varchar	小区ID
2	attempts	int(20)	尝试次数
3	accepts	int(20)	接受次数
4	succRatio	int(20)	尝试成功率
5	trafficUL	int(20)	上行流量
6	trafficDL	int(20)	下行流量
7	totalTraffic	int(20)	总流量
8	retranUL	int(20)	重传上行报文数
9	retranDL	int(20)	重传下行报文数
10	retranTraffic	int(20)	重传报文数据
11	failCount	int(20)	延时失败次数
12	transDelay	int(20)	传输时延

2. 建表语句

create table D_H_HTTP_CELLID(hourid datetime, cellid varchar(20), attempts bigint, accepts bigint, succratio bigint, trafficul bigint, trafficdl bigint, totaltraffic bigint, retranul bigint, retrandl bigint, retrantraffic bigint, failcount bigint, transdelay bigint);

3. 插入语句

insert into D_H_HTTP_CELLID select reporttime as hourid, cellid, sum(attempts) as attempts, sum(accepts) as accepts, sum(accepts)/sum(attempts) as succRatio, sum(trafficul) as trafficul, sum(trafficdl) as trafficdl, sum(trafficul)+sum(trafficdl) as totaltraffic, sum(retranul) as retranul, sum(retrandl) as retrandl, sum(retranul)+sum(retrandl) as retrantraffic, sum(failcount) as failcount, sum(transdelay) as transdelay from F_HTTP_APP_HOST group by reporttime, cellid;

3. 小区上网喜好

1. 字段

序号	字段	字段类型	描述
0	hourid	datetime	小时时间片
1	cellid	varchar	小区ID
2	host	varchar(50)	域名
3	attempts	int(20)	尝试次数
4	accepts	int(20)	接受次数
5	succRatio	int(20)	尝试成功率
6	trafficUL	int(20)	上行流量
7	trafficDL	int(20)	下行流量
8	totalTraffic	int(20)	总流量
9	retranUL	int(20)	重传上行报文数
10	retranDL	int(20)	重传下行报文数
11	retranTraffic	int(20)	重传报文数据
12	failCount	int(20)	延时失败次数
13	transDelay	int(20)	传输时延

2. 建表语句

create table D_H_HTTP_CELLID_HOST(hourid datetime, cellid varchar(20), host varchar(255), attempts bigint, accepts bigint, succratio bigint, trafficul bigint, trafficdl bigint, totaltraffic bigint, retranul bigint, retrandl bigint, retrantraffic bigint, failcount bigint, transdelay bigint);

3. 插入语句

insert into D_H_HTTP_CELLID_HOST select reporttime as hourid, cellid, host, sum(attempts) as attempts, sum(accepts) as accepts, sum(accepts)/sum(attempts) as succRatio, sum(trafficul) as trafficul, sum(trafficdl) as trafficdl, sum(trafficul)+sum(trafficdl) as totaltraffic, sum(retranul) as retranul, sum(retrandl) as retrandl, sum(retranul)+sum(retrandl) as retrantraffic, sum(failcount) as failcount, sum(transdelay) as transdelay from F_HTTP_APP_HOST group by reporttime, cellid, host;

4. 查询语句

1. 查询欢迎度前十的应用。

select apptype, DATE_FORMAT(hourid, '%Y%m%d') dateid, sum(totalTraffic) from D_H_HTTP_APPTYPE group by apptype, dateid having dateid ='20150615' order by sum(totaltraffic) desc limit 0, 10;

2. 子应用欢迎度前10。

select apptype, appsubtype, DATE_FORMAT(hourid, '%Y%m%d') dateid, sum(totalTraffic) from D_H_HTTP_APPTYPE group by apptype, appsubtype, dateid having dateid='20150615' and apptype=15 order by sum(totaltraffic) desc limit 0, 10;

3. 网站表现前10

select host, DATE_FORMAT(hourid, '%Y%m%d') dateid, sum(attempts) from D_H_HTTP_HOST group by host, dateid having dateid='20150615' order by sum(attempts) desc limit 0, 10;

4. 单网站一天表现

select host, hourid, sum(attempts) from D_H_HTTP_HOST group by host, hourid having host='apilocate.amap.com' order by attempts desc;

5. 总小区上网能力

select DATE_FORMAT(hourid, '%Y%m%d') dateid, cellid, sum(totaltraffic) from D_H_HTTP_CELLID group by dateid, cellid having dateid='20150615' limit 0, 10;

6. 指定小区上网能力

select hourid, cellid, sum(totaltraffic) from D_H_HTTP_CELLID group by cellid, hourid having DATE_FORMAT(hourid, '%Y%m%d')='20150615' and cellid='131432478';