11.2 数据探索与预处理

11.2数据探索与预处理主讲：李强任务描述业务数据表有用户基本信息表、账单信息表、订单信息表、用户状态信息变更表以及用户收视行为信息表，本小节将对这5张业务数据表进行总体的分析概述、数据探索与数据预处理。任务分析广电业务数据表的数据探索与预处理操作步骤如下。（1）数据总体概述。（2）异常数据探索。（3）主要业务数据探索。（4）标签阈值探索。（5）数据预处理。11.2.1数据总体概述1．统计用户数与记录数启动Spark集群，执行命令“spark-shell”打开SparkShell，读取Hive中的数据，如代码

11-2所示。代码11-2读取数据valhiveContext=neworg.apache.spark.sql.hive.HiveContext(sc)valusermsgData=hiveContext.sql("select*fromuser_profile.mediamatch_usermsg")valusereventData=hiveContext.sql("select*fromuser_profile.mediamatch_userevent")valbilleventsData=hiveContext.sql("select*fromuser_profile.mmconsume_billevents")valorderData=hiveContext.sql("select*fromuser_profile.order_index")valmediaData=hiveContext.sql("select*fromuser_profile.media_index")为了初步了解每个表的数据量及用户数量，现统计每个表的记录数及用户数，如代码11-3所示。代码11-3统计记录数、用户数usermsgData.countusereventData.countbilleventsData.countorderData.countmediaData.countusermsgData.select("phone_no").distinct().count2．统计观看时长为了掌握用户收视行为记录中的观看时长的取值范围，统计用户收视记录观看时长的均值、最值和标准差，如代码11-4所示。观看时长情况如图11-7所示。代码11-4统计用户收视记录观看时长的均值、最值和标准差mediaData.select(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统计每个用户月均收视时长valperUserDuraton=mediaData.groupBy("phone_no").agg((sum("duration")/(3*1000*60*60)).alias("duration_avg"))perUserDuraton.orderBy(col("duration_avg").desc).show(5)图11-8每个用户每个月的观看时长11.2.2异常数据探索1．查看用户信息表是否存在重复记录的用户通过groupBy进行分组，count统计重复数据，filter查看记录数大于1的用户，如代码11-6所示。用户记录数如图11-9所示。图11-9用户的记录数代码11-6探索用户基本信息表中重复记录的用户usermsgData.groupBy("phone_no").count().filter("count>1").countusermsgData.groupBy("phone_no").count().orderBy(col("count").desc).show(3)2．查看特殊线路的用户根据客户业务人员提供的信息，owner_code的值为02、09或10的记录是特殊路线的用户，接下来对这5个表是否存在特殊线路的用户及数量进行探索分析，如代码11-7所示。用户线路类别如图11-10所示。代码11-7统计各表owner_code类型数valusermsgCode=usermsgData.groupBy("owner_code").count()valusereventCode=usereventData.groupBy("owner_code").count()valbilleventsCode=billeventsData.groupBy("owner_code").count()valorderCode=orderData.groupBy("owner_code").count()valmediaCode=mediaData.groupBy("owner_code").count()usermsgCode.show(20)图11-10用户的线路类别3．查看政企用户该企业的客户主要来自于家庭用户，所以政企用户不纳入分析范围。政企用户的标识是owner_name的值为“EA级”“EB级”“EC级”“ED级”或“EE级”，探索这些表是否有存在政企用户以及存在的数量，如代码11-8所示。代码11-8统计用户基本信息中owner_name的分组信息valusermsgOwnerName=usermsgData.groupBy("owner_name").count()valusereventOwnerName=usereventData.groupBy("owner_name").count()valbilleventsOwnerName=billeventsData.groupBy("owner_name").count()valorderOwnerName=orderData.groupBy("owner_name").count()valmediaOwnerName=mediaData.groupBy("owner_name").count()usermsgOwnerName.show(20)usereventOwnerName.show(20)billeventsOwnerName.show(20)mediaOwnerName.show(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的分组信息11.2.3主要业务数据探索1．统计基本信息表的类型对应的用户数及其占比统计用户基本信息表中sm_name的所有类型及其每种类型的用户数，如代码11-9所示。品牌类型对应的用户数及占比如图11-15所示。代码11-9统计用户基本信息表的sm_name各类型的数量valnums=usermsgData.countusermsgData.groupBy("sm_name").count().withColumn("percent",col("count")/nums).show图11-15sm_name类型的数量与占比2．用户状态名称过滤对用户状态名称利用groupBy、count进行分组统计，如代码11-10所示。用户状态各类别记录数如图11-16所示。状态名称包含空值一共有8种类型，其中只保留正常、欠费暂停、主动暂停和主动销户的用户，其余的状态名称不需要分析处理。代码11-10统计用户基本信息表中用户状态各类别的记录数usermsgData.groupBy("run_name").count().show()图11-16用户状态记录数3．用户收视行为无效数据用户收视行为信息表记录了用户的每次收视时长，对于时长过短以及过长的数据记录都是无效的，过短是因为用户切换频道时产生的记录，过长可能是因为用户没有关闭机顶盒造成的记录，但过短与过长的范围需要进行统计分析得到。（1）在用户收视行为信息表中，duration字段记录了用户的每次收视时长，把观看长以每小时为一区间来划分，统计各区间的记录数，如代码11-11所示。观看时长的记录数及占比如图11-17所示。代码11-11统计收视时长以每小时为区间的记录数valtotal=mediaData.count.toDoublevalmediaHours=mediaData.withColumn("hours",floor(col("duration")/(1000*60*60)))valhoursNum=mediaHours.groupBy("hours").count().withColumn("percent",col("count")/total)hoursNum.show图11-17观看时长的记录数与占比3．用户收视行为无效数据由图11-17所示结果，可知由于观看时长小于1小时的记录数占了绝大部分，因此把这部分记录再按每1分钟为一个时间间隔来划分，分析落在每个区间的记录数分布情况，如代码

11-12所示。观看时长小于一小时的各区间记录数及占比如图11-18所示代码11-12统计收视行为中观看时长小于1小时的各区间记录数valoneHour=mediaHours.where("hours=0")valmediaMinutes=oneHour.withColumn("minutes",floor(col("duration")/(1000*60)))valminutesNum=mediaMinutes.groupBy("minutes").count().withColumn("percent",col("count")/total)minutesNum.orderBy(col("minutes").asc).show(10)图11-18小于一小时的各区间记录数及占比3．用户收视行为无效数据（2）由图11-18可知，观看时长小于1分钟的记录数约占总记录数的18%。为了进一步了解观看时长小于1分钟的秒级分布情况，再把这部分数据按每秒的区间间隔来划分，如代码11-13所示。观看时长小于1分钟的各区间记录数及占比如图11-19所示。代码11-13统计用户收视行为中小于1分钟各区间的记录数//筛选观看时长小于1分钟的记录valoneMinute=mediaMinutes.where("minutes=0")//新增字段seconds：将观看时长转为以每秒为间隔的数并向下取整valmediaSeconds=oneMinute.withColumn("seconds",floor(col("duration")/1000))//统计各seconds值的记录数valsecondsNum=mediaSeconds.groupBy("seconds").count().withColumn("percent",col("count")/total)//按字段seconds升序排序并显示前10条结果secondsNum.orderBy(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查询收视记录中的无效数据valinvalidData=mediaData.filter("res_type=0").filter(col("origin_time").endsWith("00")andcol("end_time").endsWith("00"))invalidData.countinvalidData.show(2,false)图11-21无效的收视纪录图11-20无效收视纪录的数量11.2.4标签阈值探索1．电视用户账单分析本小节将通过对账单信息数据中的电视用户的账单信息进行分析，得到一个电视用户的消费水平标签划分规则。（1）对电视用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费金额作基本的统计分析，如代码11-15所示。电视用户月均消费情况如图11-22所示。代码11-15统计电视用户月均消费情况valtvBilleventsData=billeventsData.filter("sm_namelike'%电视%'andsm_name!='模拟有线电视'")valavgTVBillData=tvBilleventsData.groupBy("phone_no").agg((sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))//统计消费金额的最大值，最小值，平均值和标准差avgTVBillData.select(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的消费区间，这是因为大部分电视用户每月都只缴纳26.5元基本电视费用。筛选电视用户月均消费大于等于-10且小于等于90元的用户数据，统计其平均值和标准差，如代码11-17所示。代码11-16金额划分后电视用户月均消费情况valrangeTVBillData=avgTVBillData.withColumn("range_fee",floor(col("avg_fee")/10)*10)valrangeTVCount=rangeTVBillData.groupBy("range_fee").count()rangeTVCount.orderBy(col("range_fee").asc).show(30)代码11-17过滤后的电视用户月均消费金额及标准差avgTVBillData.filter("avg_fee>=-10andavg_fee<=90").select(avg("avg_fee"),stddev("avg_fee")).show图11-23按10划分的月均消费情况1．电视用户账单分析过滤后电视用户月均消费情况如图11-24所示。从代码11-17的统计结果发现，月均消费金额在-10到90之间的电视用户月均消费金额的平均值约为26.3，标准差约为16.2。根据业务特点，结合月均消费金额的平均值和标准差，标签值以月度消费26.3为基础，以标准差的向上取整至十位数，即20作为浮动阈值，制定4个电视消费水平子标签内容如表11-7所示。代码11-17过滤后的电视用户月均消费金额及标准差avgTVBillData.filter("avg_fee>=-10andavg_fee<=90").select(avg("avg_fee"),stddev("avg_fee")).show表11-7电视消费水平标签图11-24过滤后的月均消费金额及标准差父级标签子标签标签规则备注电视消费水平电视超低消费X<26.3X为电视用户月均消费金额，单位：元电视低消费26.3≤X＜26.3+20电视中等消费26.3+20≤X＜26.3+40电视高消费26.3+40≤X2．宽带用户的账单分析本小节将通过对账单信息数据中的宽带用户的账单信息进行分析，得到一个宽带用户的消费水平标签划分规则。（1）对宽带用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费金额作基本的统计分析，如代码11-18所示。代码11-18统计宽带用户月均消费情况valnetBilleventsData=billeventsData.filter("sm_namelike'%珠江宽频%'")valavgNetBillData=netBilleventsData.groupBy("phone_no").agg((sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))importmons.math3.stat.descriptive.rank.Percentileval_50_P=newPercentile(50.0)val_50_udf=udf{(arr:scala.collection.mutable.WrappedArray[Double])=>_50_P.evaluate(arr.sorted.toArray)}avgNetBillData.select(max("avg_fee"),min("avg_fee"),avg("avg_fee"),stddev("avg_fee"),_50_udf(collect_list("avg_fee")).alias("median")).show2．宽带用户的账单分析本小节将通过对账单信息数据中的宽带用户的账单信息进行分析，得到一个宽带用户的消费水平标签划分规则。（1）对宽带用户的账单数据进行探索分析，统计每个用户的月均消费金额，然后对所有用户的月均消费金额作基本的统计分析，如代码11-18所示。代码11-18统计宽带用户月均消费情况valnetBilleventsData=billeventsData.filter("sm_namelike'%珠江宽频%'")valavgNetBillData=netBilleventsData.groupBy("phone_no").agg((sum(col("should_pay")-col("favour_fee"))/7).alias("avg_fee"))importmons.math3.stat.descriptive.rank.Percentileval_50_P=newPercentile(50.0)val_50_udf=udf{(arr:scala.collection.mutable.WrappedArray[Double])=>_50_P.evaluate(arr.sorted.toArray)}avgNetBillData.select(max("avg_fee"),min("avg_fee"),avg("avg_fee"),stddev("avg_fee"),_50_udf(collect_list("avg_fee")).alias("median")).show2．宽带用户的账单分析宽带用户月均消费情况如图11-25所示。（2）将每月消费金额按10的间隔划分，通过range_fee分组，如代码11-19所示。代码11-19分隔电视用户月均消费情况valrangeNetBillData=avgNetBillData.withColumn("range_fee",floor(col("avg_fee")/10)*10)valrangeNetCount=rangeNetBillData.groupBy("range_fee").count()rangeNetCount.orderBy(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过滤后宽带用户月均消费金额的统计avgNetBillData.filter("avg_fee>=0andavg_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<29Y为宽带用户月均消费金额，单位：元宽带中消费29≤Y＜29+19宽带高消费29+19≤Y3．电视入网程度标签阙值探索本小节将通过对用户基本信息表中的电视用户的开户时间信息进行分析，得到一个电视用户的入网程度标签划分规则。（1）把用户基本信息表的open_time字段的值与当前时间作差并把差值转为以年为单位，然后统计所有用户中入网时长的最值、均值、30%分位数和中位数，具体代码如代码11-21所示。代码11-21统计分析电视用户入网时长valtvUsermsgData=usermsgData.filter("sm_namelike'%电视%'andsm_name!='模拟有线电视'")//过滤open_time为空值的记录valtvFilteredUsermsgData=tvUsermsgData.filter("open_time!='NULL'")//把用户开户时间与当前时间作差并向下取整（单位为年）valyearTVUserData=tvFilteredUsermsgData.groupBy("phone_no").agg(max(floor(datediff(current_date(),col("open_time"))/365)).alias("years"))val_30_P=newPercentile(30.0)val_30_udf=udf{(arr:scala.collection.mutable.WrappedArray[Double])=>_30_P.evaluate(arr.sorted.toArray)}val_50_P=newPercentile(50.0)......省略3．电视入网程度标签阙值探索电视入网时长情况如图11-28所示。（2）通过years分组统计每个入网户时长值的用户数，如代码11-22所示。代码11-22统计电视每个入网户时长值valyearTVUserCount=yearTVUserData.groupBy("years").countyearTVUserCount.orderBy(col("years").asc).show(30)图11-28电视入网时长3．电视入网程度标签阙值探索电视入网时长用户合计数情况如图11-29所示。（3）根据代码11-21和代码11-22的统计结果可以发现，电视用户入网时长的平均值约为11.7，标准差约为1.16，30%分位数约为12，中位数约为12。根据统计结果及结合实际的业务场景，选择以30%分位数11年为电视用户的入网时长的中位数，以标准差取整为2年浮动值，制定三个电视入网程度子标签如表11-9所示。图11-29入网时长的合计数表11-9电视入网程度标签父级标签子标签标签规则备注电视入网程度新用户Y<9Y为用户入网时长，单位：年中等用户9≤Y＜13老用户13≤Y4．宽带入网程度标签阙值探索在对宽带用户的入网时长统计分析中，主要统计分析宽带用户中入网时长的最值、均值、标准差、中位数及其分布情况，如代码11-23所示。代码11-23统计分析宽带用户入网时长//筛选宽带用户valnetUsermsgData=usermsgData.filter("sm_name='珠江宽频'")//过滤open_time为空值的记录valnetFilteredUsermsgData=netUsermsgData.filter("open_time!='NULL'")//把用户开户时间与当前时间作差并向下取整（单位为年）valyearNetUserData=netFilteredUsermsgData.groupBy("phone_no").agg(max(floor(datediff(current_date(),col("open_time"))/365)).alias("years"))//统计入网时长的最大值，最小值，均值，标准差，中位数yearNetUserData.select(max("years"),min("years"),avg("years"),stddev("years"),_50_udf(collect_list(col("years").cast("double"))).alias("median")).show4．宽带入网程度标签阙值探索宽带入网时长情况如图11-30所示。以years字段分组，统计每个宽带入网户时长值的用户数，如代码11-24所示。宽带入网时长用户合计数情况如图11-31所示。代码11-24统计每个宽带入网时长的用户数valyearNetUserCount=yearNetUserData.groupBy("years").countyearNetUserCount.orderBy(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<8Y为用户入网时长,单位：年中等用户8≤Y＜14老用户14≤Y11.2.5数据预处理参考项目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.2.5数据预处理从表11-11可以发现这5个表的数据预处理规则都有共同的地方，此外用户信息表与用户收视行为信息表还有其他的处理规则。根据此特点，可以封装成一个函数来实现数据处理过程，以达到代码复用的目的，此函数需要考虑以下参数的问题。①需要传入一个HiveContext实例，因为需要在Hive中进行读取和写入的操作；②需要指定从Hive读取数据的表名称；③由于不同的表的处理逻辑不一样，所以需要一个标记参数区分输入的表；④数据预处理完毕后需要指定数据存储在Hive的表名称。根据以上的业务逻辑及参数封装，得到关于数据预处理功能的代码，如代码11-25所示。代码11-25数据预处理/***数据预处理*/objectDataProcess{defmain(args:Array[String]):Unit={if(args.length!=10){printUsage()System.exit(1)}valconf=newSparkConf().setAppName("DataProcess")valsc=newSparkContext(conf)valsqlContext=newHiveContext(sc)//media_index数据预处理valoriginMediaIndexTable=args(0)valprocessMediaIndexTable=args(1)11.2.5数据预处理（2）参考项目3，将打包成datapro.jar，然后将jar包上传至liunx的opt目录下，执行命令“cd/opt”进入jar包所在目录，运行spark-submit测试代码，命令如代码11-26所示。代码11-26运行预处理任务spark-submit--masterspark://master:7077--total-ex