Spark DataFrameRDD案例实现

Spark DataFrame&RDD案例实现

1.DataFram类型
2.需求：将orders表中days_since_prior_order中的空置(" ")转换成0
idea中快捷键
2.每个用户平均购买订单的间隔周期
3.每个用户的总订单数量（分组）
4.每个用户购买的product商品去重后的集合数据
4.每个用户总商品数量以及去重后的商品数量(distinct count)
需求：合并“去重后集合的数据”和“去重后的商品数量” 统计逻辑
5.每个用户购买的平均每个订单的商品数量（hive已经实现过了）

1.DataFram类型

针对NULL值可以转换成0

df.na.fill(0)

2.需求：将orders表中days_since_prior_order中的空置(" ")转换成0

val orders = spark.sql("select * from badou.orders")
val priors = spark.sql("select * from badou.priors")orders表
+--------+-------+--------+------------+---------+-----------------+----------------------+
|order_id|user_id|eval_set|order_number|order_dow|order_hour_of_day|days_since_prior_order|
+--------+-------+--------+------------+---------+-----------------+----------------------+
| 2539329|      1|   prior|           1|        2|               08|                      |
| 2398795|      1|   prior|           2|        3|               07|                  15.0|
|  473747|      1|   prior|           3|        3|               12|                  21.0|
| 2254736|      1|   prior|           4|        4|               07|                  29.0|
|  431534|      1|   prior|           5|        4|               15|                  28.0|
+--------+-------+--------+------------+---------+-----------------+----------------------+

因为days_since_prior_order中有空值，所以要进行空值填充，可以利用selectExpr()

scala> orders.selectExpr("*","if(days_since_prior_order='',0.0 ,days_since_prior_order) as dspo").show(5)
+--------+-------+--------+------------+---------+-----------------+----------------------+----+
|order_id|user_id|eval_set|order_number|order_dow|order_hour_of_day|days_since_prior_order|dspo|
+--------+-------+--------+------------+---------+-----------------+----------------------+----+
| 2539329|      1|   prior|           1|        2|               08|                      | 0.0|
| 2398795|      1|   prior|           2|        3|               07|                  15.0|15.0|
|  473747|      1|   prior|           3|        3|               12|                  21.0|21.0|
| 2254736|      1|   prior|           4|        4|               07|                  29.0|29.0|
|  431534|      1|   prior|           5|        4|               15|                  28.0|28.0|
+--------+-------+--------+------------+---------+-----------------+----------------------+----+
only showing top 5 rows

删除days_since_prior_order这列

scala> val orderNew = orders.selectExpr("*","if(days_since_prior_order='',0.0,days_since_prior_order) as dspo").drop("days_since_prior_order")
orderNew: org.apache.spark.sql.DataFrame = [order_id: string, user_id: string ... 5 more fields]scala> orderNew.show(5)
+--------+-------+--------+------------+---------+-----------------+----+
|order_id|user_id|eval_set|order_number|order_dow|order_hour_of_day|dspo|
+--------+-------+--------+------------+---------+-----------------+----+
| 2539329|      1|   prior|           1|        2|               08| 0.0|
| 2398795|      1|   prior|           2|        3|               07|15.0|
|  473747|      1|   prior|           3|        3|               12|21.0|
| 2254736|      1|   prior|           4|        4|               07|29.0|
|  431534|      1|   prior|           5|        4|               15|28.0|
+--------+-------+--------+------------+---------+-----------------+----+

idea中快捷键

将多行合并成一场 ctrl shift + j
返回上一步操作： ctrl + z

2.每个用户平均购买订单的间隔周期

注意：用户的第一个订单没有间隔天数的，需要赋值为0间隔天数在orders表中

scala> val userGap = orderNew.selectExpr("user_id","cast(dspo as int) as dspo").groupBy("user_id").avg("dspo")scala> userGap.show(5)+-------+------------------+
|user_id|         avg(dspo)|
+-------+------------------+
|    296| 5.428571428571429|
|    467| 8.833333333333334|
|    675|              20.0|
|    691|13.173913043478262|
|    829|              9.25|
+-------+------------------+
only showing top 5 rows

将avg(dspo)进行重命名

scala> val userGap = orderNew.selectExpr("user_id","cast(dspo as int) as dspo").groupBy("user_id").avg("dspo").withColumnRenamed("avg(dspo)","u_avg_day_gap")

3.每个用户的总订单数量（分组）

scala> val userOrdCnt = orders.groupBy("user_id").count()scala> userOrdCnt.show(5)+-------+-----+
|user_id|count|
+-------+-----+
|    296|    7|
|    467|    6|
|    675|   11|
|    691|   23|
|    829|    4|
+-------+-----+
only showing top 5 rows

4.每个用户购买的product商品去重后的集合数据

结果： 1001 101200,120219,129101
将orders与priors表进行关联
问题：""代表String ’ '代表char类型

scala> val opDF = orders.join(priors,"order_id")
scala> val up = opDF.select("user_id","product_id")

将DataFrame 转变成RDD需要下面隐式转换包，利用隐式转换因为后面Spark的算子操作如map()、groupBy()

DataFrame中没有mapValues()！！！！！！
注意：在使用RDD算子前进行(.rdd)转换就行

import spark.implicits._

注意：RDD类型没有show() 只有take()来查看数据

a、将product_id按照user_id进行分组
groupByKey() 按key进行分组注意键值对 => {key:values}

方式一：
scala> val rddGroup = up.rdd.map{x=>(x(0).toString,x(1).toString)}.groupByKey()
方式二：
scala> val rdGroup = up.map{ x=>(x(0).toString,x(1).toString)}.rdd.groupByKey()
结果：take(5) //从RDD中获取5个元素
scala> rddGroup.take(5):36 WARN executor.Executor: Managed memory leak detected; size = 5244782 bytes, TID = 415
res11: Array[(String, Iterable[String])] = Array((124168,CompactBuffer(20082, 20082, 14303, 11323, 46522, 11323, 20082, 20082, 22108, 22108))
, (120928,CompactBuffer(39275, 18897, 27845, 47209, 30440, 34448, 45948, 4605, 27966, 30252, 20995))

b、将商品进行去重

scala> mapValues(record=>record.toSet.mkString(","))   product商品去重(toSet)后的利用mkString(",")逗号进行拼接
var rddRecords = up.rdd.map{x=>(x(0).toString,x(1).toString)}.groupByKey().mapValues(record=>record.toSet.mkString(","))
结果
scala> rddRecords.take(5)
:18 WARN executor.Executor: Managed memory leak detected; size = 5244782 bytes, TID = 214
res9: Array[(String, String)] = Array((124168,20082,20082,14303,11323,46522,11323,20082,20082,22108,22108)
, (120928,39275,18897,27845,47209,30440,34448,45948,4605,27966,30252,20995,5194,13629,36695,25824,42265,39928,13870,41665,39561,4799)

d、RDD 转换成DF
toDF(“自定义名1”,“自定义名2”) 根据RDD已有的列数可以进行重命名

scala> rddRecords.toDF("user_id","prod_cnt").show(5)
结果
size = 5244782 bytes, TID = 423
+-------+--------------------+
|user_id|            prod_cnt|
+-------+--------------------+
| 124168|20082,20082,14303...|
| 120928|39275,18897,27845...|
| 186692|11365,6184,14161,...|
|  43535|38185,33129,13187...|
|   8965|33198,25466,27966...|
+-------+--------------------+

4.每个用户总商品数量以及去重后的商品数量(distinct count)

a、user所有的商品数量(没有去重)

val  userAllProd = up.groupBy("user_id").count()
scala> userAllProd.show(5)
+-------+-----+
|user_id|count|
+-------+-----+
|  88447|  185|
| 144913|  198|
| 145079|  612|
|  13282|   99|
| 124057|   80|
+-------+-----+
only showing top 5 rows

b、去重后的商品数量

val userUnOrdCnt = up.rdd.map{x=>(x(0).toString, x(1).toString)}.groupByKey().mapValues(_.toSet.size).toDF("user_id","prod_dis_cnt")

汇总关联：

scala> userAllProd.join(userUnOrdCnt,"user_id").select("*").show(10)
+-------+-----+------------+
|user_id|count|prod_dis_cnt|
+-------+-----+------------+
| 100010|  206|         119|
| 100140|   32|          28|
| 100227|  134|          70|
| 100263|  212|          38|
| 100320|  219|         121|
| 100553|  645|         243|
| 100704|   50|          40|
| 100735|   50|          42|
| 100768|    9|           8|
|  10096|   90|          47|
+-------+-----+------------+
only showing top 10 rows

//处理groupByKey的时候，看看能不能进行提前合并

需求：合并“去重后集合的数据”和“去重后的商品数量” 统计逻辑

方式一：合并提取公因子，引入cache 提高效率

val userRddGroup = up.rdd.map{x=>(x(0).toString, x(1).toString)}.groupByKey().cache()userRddGroup.mapValues(_.toSet.mkString(",")).toDF("user_id","prod_records")
从缓存中移除  python  del xxx
userRddGroup.unpersist()

方式二:同时计算两个

//返回tuple类型 (1.商品的大小(size)，2.商品的List集合，利用逗号进行拼接)
mapValues{records=>
val rs = records.toSet;    (rs.size , rs.mkString(","))}
val userProRcdSize = up.rdd.map{x=>(x(0).toString, x(1).toString)}.groupByKey().mapValues{records=>val rs = records.toSet;//返回tuple类型 (1.商品的大小(size)，2.商品的List集合，利用逗号进行拼接)(rs.size , rs.mkString(","))}.toDF("user_id", "tuple").selectExpr("user_id","tuple._1 as prod_dist_size", "tuple._2 as prod_records")

方式三：使用自带的函数的处理
as(“af”) 重命名可以使用 withColumnRenamed()替代但是位置不同

//导入sql中所有的包
import org.apache.spark.sql.functions._val usergroup = up.groupBy("user_id").agg(size(collect_set("product_id")).as("prod_dist_size"),collect_set("product_id").as("prod_records"))
结果：
+-------+--------------+--------------------+
|user_id|prod_dist_size|        prod_records|
+-------+--------------+--------------------+
| 100010|           119|[7751, 27360, 247...|
| 100140|            28|[7021, 27845, 436...|
| 100227|            70|[24834, 45007, 27...|
| 100263|            38|[5157, 38928, 372...|
| 100320|           121|[39891, 34358, 17...|
| 100553|           243|[6873, 46075, 205...|
| 100704|            40|[14999, 3434, 308...|
| 100735|            42|[45368, 7430, 207...|
| 100768|             8|[40199, 18838, 49...|
|  10096|            47|[8021, 13629, 895...|
+-------+--------------+--------------------+
only showing top 10 rows

5.每个用户购买的平均每个订单的商品数量（hive已经实现过了）

我今天购买了2个order，一个是10个商品，另一个是4个商品
（10+4）一个订单对应多少个商品 / 2
结果：一个用户购买了几个商品=7

//    1) 每个订单有多少个商品
val ordProdCnt = priors.groupBy("order_id").count()//    2)求每个用户订单商品数量的平均值  user_id  product_id
val userPerOrdProdCnt = orders.join(ordProdCnt, "order_id").groupBy("user_id").agg(avg("count").as("u_avg_ord_prods"))
结果：
scala> userPerOrdProdCnt.show(5)
+-------+---------------+
|user_id|u_avg_ord_prods|
+-------+---------------+
|  88447|            7.4|
| 144913|            5.5|
| 145079|            8.5|
|  13282|         6.1875|
| 124057|           20.0|
+-------+---------------+
only showing top 5 rows