数据探索与预处理
主讲：李强
任务描述
业务数据表有用户基本信息表、账单信息表、订单信息表、用户状态信息变更
表以及用户收视行为信息表，本小节将对这5张业务数据表进行总体的分析概
述、数据探索与数据预处理。
任务分析
广电业务数据表的数据探索与预处理操作步骤如下。
（1）数据总体概述。
（2）异常数据探索。
（3）主要业务数据探索。
（4）标签阈值探索。
（5）数据预处理。
数据总体概述
1．统计用户数与记录数
启动Spark集群，执行命令“spark-shell”打开Spark Shell，读取Hive中的数据，如代码 11-2所示。
代码11-2 读取数据
val hiveContext = new (sc)
val usermsgData = ("select * from _usermsg")
val usereventData = ("select * from _userevent")
val billeventsData = ("select * from _billevents")
val orderData = ("select * from _index")
val mediaData = ("select * from _index")
为了初步了解每个表的数据量及用户数量，现统计每个表的记录数及用户数，如代码11-3所示。
代码11-3 统计记录数、用户数
("phone_no").distinct().count
2．统计观看时长
为了掌握用户收视行为记录中的观看时长的取值范围，统计用户收视记录观看时长的均值、最值和标准差，如代码11-4所示。
观看时长情况如图11-7所示。
代码11-4 统计用户收视记录观看时长的均值、最值和标准差
(avg(col("duration")/1000).alias("avg_duration"),min(col("duration")/1000)
.alias("min_duration"),max(col("duration")/1000).alias("max_duration"),stddev(col("duration")/1000)
.alias("std_duration")).show
图11-6 表的记录数和用户数
图11-7 观看时长的均值、最值和标准差
2．统计观看时长
统计每个用户平均每月的收视时长，进而掌握每个用户对电视的依赖程度，如代码11-5所示。
代码11-5 统计每个用户月均收视时长
val perUserDuraton = ("phone_no").agg((sum("duration")/(3*1000*60*60))
.alias("duration_avg"))
(col("duration_avg").desc).show(5)
图11-8 每个用户每个月的观看时长
异常数据探索
1．查看用户信息表是否存在重复记录的用户
通过groupBy进行分组，count统计重复数据，filter查看记录数大于1的用户，如代码11-6所示。用户记录数如图11-9
所示。
图11-9 用户的记录数
代码11-6 探索用户基本信息表中重复记录的用户
("phone_no").count().filter("count>1").count
("phone_no").count().orderBy(col("count").desc).show(3)
2．查看特殊线路的用户
根据客户业务人员提供的信息，owner_code的值为02、09或10的
记录是特殊路线的用户，接下来对这5个表是否存在特殊线路的用
户及数量进行探索分析，如代码11-7所示。用户线路类别如图11-
10所示。
代码11-7 统计各表owner_code类型数
val usermsgCode = ("owner_code").count()
val usereventCode = ("owner_code").count()
val billeventsCode = ("owner_code").count()
val orderCode = ("owner_code").count()
val mediaCode = ("owner_code").count()
(20)
图11-10 用户的线路类别
3．查看政企用户
该企业的客户主要来自于家庭用户，所以政企用户不纳入分析范围。政企用户的标识是owner_name的值为
“EA级”“EB级”“EC级”“ED级”或“EE级”，探索这些表是否有存在政企用户以及存在的数量，如代码11-8所示。
代码11-8 统计用户基本信息中owner_name的分组信息
val usermsgOwnerName = ("owner_name").count()
val usereventOwnerName = ("owner_name").count()
val billeventsOwnerName = ("owner_name").count()
val orderOwnerName = ("owner_name").count()
val mediaOwnerName = ("owner_name").count()
(20)
(20)
(20)
(20)
3．查看政企用户
用户基本信息表owner_name的分组信息如图11-11所示。
用户状态信息变更表owner_name的分组信息如图11-12所示。
图11-11 用户基本信息表owner_name的分组信息 图11-12 用户状态信息变更表owner_name的分组信息
3．查看政企用户
账单信息表owner_name的分组信息如图11-13所示。
用户收视行为信息表owner_name的分组信息如图11-14所示。
图11-13 账单信息表owner_name的分组信息 图11-14 用户收视行为信息表表owner_name的分组信息
主要业务数据探索
1．统计基本信息表的类型对应的用户数及其占比
统计用户基本信息表中sm_name的所有类型及其每种类型的用户数，如代码11-9所示。
品牌类型对应的用户数及占比如图11-15所示。
代码11-9 统计用户基本信息表的sm_name各类型的数量
val nums =
("sm_name").count().withColumn("percent",col("count")/nums).show
图11-15 sm_name类型的数量与占比
2．用户状态名称过滤
对用户状态名称利用groupBy、count进行分组统计，如代
码11-10所示。
用户状态各类别记录数如图11-16所示。
状态名称包含空值一共有8种类型，其中只保留正常、欠
费暂停、主动暂停和主动销户的用户，其余的状态名称不
需要分析处理。
代码11-10 统计用户基本信息表中用户状态各类别的记录数
("run_name").count().show()
图11-16 用户状态记录数
3．用户收视行为无效数据
用户收视行为信息表记录了用户的每次收视时长，对于时
长过短以及过长的数据记录都是无效的，过短是因为用户
切换频道时产生的记录，过长可能是因为用户没有关闭机
顶盒造成的记录，但过短与过长的范围需要进行统计分析
得到。
（1）在用户收视行为信息表中，duration字段记录了用户
的每次收视时长，把观看长以每小时为一区间来划分，统
计各区间的记录数，如代码11-11所示。
观看时长的记录数及占比如图11-17所示。
代码11-11 统计收视时长以每小时为区间的记录数
val total=
val mediaHours= ("hours",floor(col("duration")/(1000*60*60)))
val hoursNum = ("hours").count().withColumn("percent",col("count")/total)
图11-17 观看时长的记录数与占比
3．用户收视行为无效数据
由图11-17所示结果，可知由于观看时长小于1小时的记录数占了
绝大部分，因此把这部分记录再按每1分钟为一个时间间隔来划分，
分析落在每个区间的记录数分布情况，如代码 11-12所示。
观看时长小于一小时的各区间记录数及占比如图11-18所示
代码11-12 统计收视行为中观看时长小于1小时的各区间记录数
val oneHour = ("hours=0")
val mediaMinutes = ("minutes",floor(col("duration")/(1000*60)))
val minutesNum = ("minutes").count().withColumn ("percent",col("count")/total)
(col("minutes").asc).show(10)
图11-18 小于一小时的各区间记录数及占比
3．用户收视行为无效数据
（2）由图11-18可知，观看时长小于1分钟的记录数约占总记录
数的18%。为了进一步了解观看时长小于1分钟的秒级分布情况，
再把这部分数据按每秒的区间间隔来划分，如代码11-13所示。
观看时长小于1分钟的各区间记录数及占比如图11-19所示。
代码11-13 统计用户收视行为中小于1分钟各区间的记录数
// 筛选观看时长小于1分钟的记录
val oneMinute = ("minutes=0")
// 新增字段seconds：将观看时长转为以每秒为间隔的数并向下取整
val mediaSeconds =
("seconds",floor(col("duration")/1000))
// 统计各seconds值的记录数
val secondsNum =
("seconds").count().withColumn
("percent",col("count")/total)
// 按字段seconds升序排序并显示前10条结果
(col("seconds").asc).show(10)
图11-19 小于1分钟各区间的记录数
3．用户收视行为无效数据
（3）在用户收视行为信息表中，还有一部分数据是res_type=0时，origin_time和end_time的秒时间单位为00的记录，这些记
录是机顶盒子自动返回的数据，并不是用户真实的观看记录，因此，这一部分数据在后续预处理时也是要删除的，如代码11-
14所示。
无效收视纪录数量如图11-20所示，无效的收视记录如图11-21所示。
代码11-14 查询收视记录中的无效数据
val invalidData =
("res_type=0").filter(col("origin_time")
.endsWith("00") and col("end_time").endsWith("00"))
invalidData .show(2,false)
图11-21 无效的收视纪录
图11-20 无效收视纪录的数量
标签阈值探索
1．电视用户账单分析
本小节将通过对账单信息数据中的电视用户的账单信息进行分析，得到一个电视用户的消费水平标签划分规则。
（1）对电视用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费金额作
基本的统计分析，如代码11-15所示。
电视用户月均消费情况如图11-22所示。
代码11-15 统计电视用户月均消费情况
val tvBilleventsData = ("sm_name like '%电视%' and sm_name !='模拟有线电视'")
val avgTVBillData = ("phone_no").agg(
(sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))
// 统计消费金额的最大值，最小值，平均值和标准差
(max("avg_fee"),min("avg_fee"),avg("avg_fee"),stddev("avg_fee")).show
图11-22 电视用户月均消费情况
1．电视用户账单分析
（2）将每月消费金额按10的间隔划分，通过range_fee分组，如代码11-16所示。
电视用户按10划分的月均消费情况如图11-23所示。
（3）从如图11-23所示的结果可以发现，电视用户月均消费金额大致呈正态分
布，主要集中在0至50的区间范围内，其中占比最大的是大于等于20而小于30的
消费区间，这是因为大部分电视用户每月都只缴纳元基本电视费用。筛选电
视用户月均消费大于等于-10且小于等于90元的用户数据，统计其平均值和标准
差，如代码11-17所示。
代码11-16 金额划分后电视用户月均消费情况
val rangeTVBillData = ("range_fee",floor(col("avg_fee")/10)*10)
val rangeTVCount = ("range_fee").count()
(col("range_fee").asc).show(30)
代码11-17 过滤后的电视用户月均消费金额及标准差
("avg_fee>=-10 and
avg_fee<=90").select(avg("avg_fee"),stddev("avg_fee")).show
图11-23 按10划分的月均消费情况
1．电视用户账单分析
过滤后电视用户月均消费情况如图11-24所示。
从代码11-17的统计结果发现，月均消费金额在-10到90之间
的电视用户月均消费金额的平均值约为，标准差约为
。根据业务特点，结合月均消费金额的平均值和标准差，
标签值以月度消费为基础，以标准差的向上取整至十位
数，即20作为浮动阈值，制定4个电视消费水平子标签内容如
表11-7所示。
代码11-17 过滤后的电视用户月均消费金额及标准差
("avg_fee>=-10 and
avg_fee<=90").select(avg("avg_fee"),stddev("avg_fee")).show
表11-7 电视消费水平标签
图11-24 过滤后的月均消费金额及标准差
父级标签 子标签 标签规则 备注
电视消费水
平
电视超低消
费
X<
X为电视用户月均
消费金额，单位：
元
电视低消费 ≤X＜+20
电视中等消
费
+20≤X＜
+40
电视高消费 +40≤X
2．宽带用户的账单分析
本小节将通过对账单信息数据中的宽带用户的账单信息进行分析，得到一个宽带用户的消费水平标签划分规则。
（1）对宽带用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费金额作
基本的统计分析，如代码11-18所示。
代码11-18 统计宽带用户月均消费情况
val netBilleventsData = ("sm_name like '%珠江宽频%'")
val avgNetBillData = ("phone_no").agg(
(sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))
import
val _50_P = new Percentile()
val _50_udf = udf{(arr: [Double]) =>
()}
(max("avg_fee"),min("avg_fee"),avg("avg_fee"),
stddev("avg_fee"),_50_udf(collect_list("avg_fee")).alias("median")).show
2．宽带用户的账单分析
本小节将通过对账单信息数据中的宽带用户的账单信息进行分析，得到一个宽带用户的消费水平标签划分
规则。
（1）对宽带用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费
金额作基本的统计分析，如代码11-18所示。
代码11-18 统计宽带用户月均消费情况
val netBilleventsData = ("sm_name like '%珠江宽频%'")
val avgNetBillData = ("phone_no").agg(
(sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))
import
val _50_P = new Percentile()
val _50_udf = udf{(arr: [Double]) =>
()}
(max("avg_fee"),min("avg_fee"),avg("avg_fee"),
stddev("avg_fee"),_50_udf(collect_list("avg_fee")).alias("median")).show
2．宽带用户的账单分析
宽带用户月均消费情况如图11-25所示。
（2）将每月消费金额按10的间隔划分，通过range_fee分组，如代码11-19所示。
代码11-19 分隔电视用户月均消费情况
val rangeNetBillData = ("range_fee",floor(col("avg_fee")/10)*10)
val rangeNetCount = ("range_fee").count()
(col("range_fee").asc).show
图11-25 宽带用户月均消费情况
图11-26 分隔后的电视用户月均消费情况
宽带用户按10划分后月均消费情况如图11-26所示。
2．宽带用户的账单分析
（3）宽带用户的消费金额集中在0元至90元之间，其中消费金额在10至20区间内的用户最多，为了使宽带消
费金额的均值和标准差更加稳定，过滤消费金额小于0或大于90的记录，再求其均值和标准差和中位数，如代
码11-20所示。
过滤后的宽带用户月均消费情况如图11-27所示。
代码11-20 过滤后宽带用户月均消费金额的统计
("avg_fee>= 0 and avg_fee<90").select(avg("avg_fee"),
stddev("avg_fee"),_50_udf(collect_list("avg_fee")).alias("median")).show
图11-27 过滤后的宽带用户月均消费金额
2．宽带用户的账单分析
从代码11-20的统计结果，可以发现过滤部分记录后宽带用户月均消费金额的均值约为29，标准差约为19，中
位数约为21。根据统计结果，选择29元作为宽带用户的基础消费，19元作浮动阀值，制定宽带消费水平的子标
签及规则如表11-8所示。
表11-8 宽带消费水平标签
父级标签 子标签 标签规则 备注
宽带消费水平
宽带低消费 Y<29
Y为宽带用户月均消费金额，单位：
元
宽带中消费 29≤Y＜29+19
宽带高消费 29+19≤Y
3．电视入网程度标签阙值探索
本小节将通过对用户基本信息表中的电视用户的开户时间信息进行分析，得到一个电视用户的入网程度标签划分规则。
（1）把用户基本信息表的open_time字段的值与当前时间作差并把差值转为以年为单位，然后统计所有用户中入网时
长的最值、均值、30%分位数和中位数，具体代码如代码11-21所示。
代码11-21 统计分析电视用户入网时长
val tvUsermsgData = ("sm_name like '%电视%' and sm_name !='模拟有线电视'")
// 过滤open_time为空值的记录
val tvFilteredUsermsgData = ("open_time != 'NULL'")
// 把用户开户时间与当前时间作差并向下取整（单位为年）
val yearTVUserData = ("phone_no").agg(
max(floor(datediff(current_date(),col("open_time"))/365)).alias("years"))
val _30_P = new Percentile()
val _30_udf = udf{(arr: [Double]) =>
()}
val _50_P = new Percentile()
......省略
3．电视入网程度标签阙值探索
电视入网时长情况如图11-28所示。
（2）通过years分组统计每个入网户时长值的用户数，如代码11-22所示。
代码11-22 统计电视每个入网户时长值
val yearTVUserCount=("years").count
(col("years").asc).show(30)
图11-28 电视入网时长
3．电视入网程度标签阙值探索
电视入网时长用户合计数情况如图11-29所示。
（3）根据代码11-21和代码11-22的统计结果可以发现，
电视用户入网时长的平均值约为，标准差约为，
30%分位数约为12，中位数约为12。根据统计结果及结合
实际的业务场景，选择以30%分位数11年为电视用户的入
网时长的中位数，以标准差取整为2年浮动值，制定三个
电视入网程度子标签如表11-9所示。
图11-29 入网时长的合计数
表11-9 电视入网程度标签
父级标签 子标签 标签规则 备注
电视入网程度
新用户 Y<9
Y为用户入网时长，单位：年中等用户 9≤Y＜13
老用户 13≤Y
4．宽带入网程度标签阙值探索
在对宽带用户的入网时长统计分析中，主要统计分析宽带用户中入网时长的最值、均值、标准差、中位数及其分布情况，
如代码11-23所示。
代码11-23 统计分析宽带用户入网时长
// 筛选宽带用户
val netUsermsgData = ("sm_name='珠江宽频'")
// 过滤open_time为空值的记录
val netFilteredUsermsgData = ("open_time != 'NULL'")
// 把用户开户时间与当前时间作差并向下取整（单位为年）
val yearNetUserData = ("phone_no").agg(
max(floor(datediff(current_date(),col("open_time"))/365)).alias("years"))
// 统计入网时长的最大值，最小值，均值，标准差，中位数
(max("years"),min("years"),avg("years"),stddev("years"),
_50_udf(collect_list(col("years").cast("double"))).alias("median")).show
4．宽带入网程度标签阙值探索
宽带入网时长情况如图11-30所示。
以years字段分组，统计每个宽带入网户时长值的用户数，如代码
11-24所示。
宽带入网时长用户合计数情况如图11-31所示。
代码11-24 统计每个宽带入网时长的用户数
val yearNetUserCount=("years").count
(col("years").asc).show()
图11-30 宽带入网时长
图11-31 每个宽带入网时长的用户数
4．宽带入网程度标签阙值探索
通过代码11-23和代码11-24的统计结果可以发现，宽带用户中入网时长的均值约为11，标准差约
为3，中位数为12。从中可以发现主要集中在11~14年之间，其中入网时长为13年的用户最多。
根据统计结果以及结合实际的业务场景，选择以11年作为入网时长的中间值，以标准差3年作为
浮动值，制定三个宽带入网程度子标签如表11-10所示。
表11-10 宽带入网程度标签
父级标签 子标签 标签规则 备注
宽带入网程度
新用户 Y<8
Y为用户入网时长,单位：年中等用户 8≤Y＜14
老用户 14≤Y
数据预处理
参考项目10，在IDEA中创建一个名为“zjsm”的Maven工程，并在工程里添加好Scala插件、导入Spark安装包中的
jars文件夹，配置好Spark开发环境。
通过封装函数实现数据预处理过程如下。
（1）通过前几节，得到数据清洗的规则如表11-11所示。
表11-11 业务数据表预处理规则
表名称 数据预处理规则
账单表订单表用户信息表用户收视信
息行为信息表
1．数据去重；2．删除owner_name=EA级，EB级，EC级，ED级，EE级；3．删除
owner_code=02，09，10；4．保留sm_name=“珠江宽频”“数字电视”“互动电视”“甜果电视；
用户状态信息变更表订单表用户信息
表
1．保留run_name=“正常”“主动暂停”“欠费暂停”“主动销户”；
用户信息表 1．每个用户只保留run_time时间为最大的记录；
用户收视信息行为信息表
1．收视时长duration≥20秒且duration≤5小时；2．删除用户收视行为信息表中res_type=0时
origin_time和end_time中秒单位为00的数据；
用户状态信息变更表
1．数据去重；2．删除owner_name=EA级，EB级，EC级，ED级，EE级；3．删除
owner_code=02，09，10；
数据预处理
从表11-11可以发现这5个表的数据预处理规则都有共同的
地方，此外用户信息表与用户收视行为信息表还有其他的
处理规则。根据此特点，可以封装成一个函数来实现数据
处理过程，以达到代码复用的目的，此函数需要考虑以下
参数的问题。
① 需要传入一个HiveContext实例，因为需要在Hive中进
行读取和写入的操作；
② 需要指定从Hive读取数据的表名称；
③ 由于不同的表的处理逻辑不一样，所以需要一个标记参
数区分输入的表；
④ 数据预处理完毕后需要指定数据存储在Hive的表名称。
根据以上的业务逻辑及参数封装，得到关于数据预处理功
能的代码，如代码11-25所示。
代码11-25 数据预处理
/**
* 数据预处理
*/
object DataProcess {
def main(args: Array[String]): Unit = {
if ( != 10) {
printUsage()
(1)
}
val conf = new SparkConf().setAppName("DataProcess")
val sc = new SparkContext(conf)
val sqlContext = new HiveContext(sc)
// media_index数据预处理
val originMediaIndexTable = args(0)
val processMediaIndexTable = args(1)
数据预处理
（2）参考项目3，将打包成，然后将jar包上传至liunx的opt目录下，执行命令“cd /opt”进入jar包所在目录，
运行spark-submit测试代码，命令如代码11-26所示。
代码11-26 运行预处理任务
spark-submit --master spark://master:7077 --total-executor-cores 2 \
--executor-cores 1 \
--name DataProcess \
--class \
/opt/ \
_index _index_process \
_userevent _userevent_process \
_usermsg _usermsg_process \
_billevents _billevent_process \
_index _index_process
数据预处理
（ 3）确认预处理任务运行成功后，在 Linux终端执行命令 “spark-shell”启动 spark，查看用户基本信息表
mediamatch_usermsg_process在Hive中存储情况，如代码11-27所示。
代码11-27 用户基本信息表数据预处理验证
val hiveContext = new (sc)
val originUsermsg = ("select * from _usermsg")
val processedUsermsg = ("select * from _usermsg_process")
("phone_no").
("sm_name").
("run_name").
("owner_name").
("owner_code").
数据预处理
预处理后的验证结果如图11-32、图11-33、图11-34、图11-35和图11-36所示。
图11-32 处理后表的记录数 图11-33 电视的品牌
图11-35 去除政企用户 图11-34 用户的状态变更
数据预处理
根据代码11-27的统计结果显示，可以发现根据用户基本信息表数据预处理规则过滤后其记录数减少了一大半，其记录
数和用户数也符合设置的规则（每个用户有且只有一条记录），并且其它字段的过滤如owner_name、owner_code、
sm_name、run_name等也达到了预期的效果。基于用户基本信息表的数据预处理的验证效果，说明数据预处理代码是
正确的。
图11-36 去除owner_code=02、09、10后的结果
谢谢
