spark-sql相关实践

忘是亡心i 2022-03-11 16:46 307阅读 0赞

给定数据：

orders表：

![20190308131318808.png][]

product表：

![20190308131407410.png][]

priors表：

![20190308131428643.png][]

实现以下业务需求：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70][]

### 1.统计product被购买的数量： ###

val productCnt = priors.groupBy("product\_id").count()

### 2..统计product 被reordered的数量（再次购买） ###

product\_id做group by(聚合)，统计一下sum（reorder）的值

1）priors  
.selectExpr("product\_id","cast(reordered as int)")  
.filter(col("reordered")===1)  
.groupBy("product\_id")  
.count()

2）priors.selectExpr("product\_id","cast(reordered as int)")  
.groupBy("product\_id").sum("reordered")

3）priors.selectExpr("product\_id","cast(reordered as int)")  
.groupBy("product\_id").agg(sum("reordered"),avg("reordered"))

（agg一般是搭配groupby这种聚合函数使用和单独使用sum的差别，是在一次聚合中可以统计出来多个值）

### 3.结合上面数量统计product购买的reordered 的比率avg("reordered") ###

sql解决：jproduct.selectExpr("\*","sum\_re/count as mean\_re")

udf解决：

join操作：

productCnt：为上面计算好的count

val jproduct = productCnt.join(prodcutSumReorder,"product\_id")

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1][]

使用udf：1.导包![20190308125437848.png][]

2定义udf：.val avg\_udf = udf((sm:Long,cnt:Long)=>sm.toDouble/cnt.toDouble)

3.使用udf：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 2][]

### user 统计/特征   1.每个用户平均购买订单的间隔周期 ###

1)解决用户的第一个订单没有间隔天数的，需要赋值为0

val ordersNew =orders.selectExpr("\*","if(days\_since\_prior\_order='',0,days\_since\_prior\_order) as dspo")  
.drop("days\_since\_prior\_order").show()

（drop：丢掉其中的一列）

2）最终代码：

ordersNew.selectExpr("user\_id","cast(dspo as int)")  
.groupBy("user\_id").avg("dspo").show()

### 2.每个用户的总订单数量   orders.groupBy("user\_id").count().show() ###

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

1.）先求每个用户和product：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 3][]

val po = orders.join(priors,"order\_id").select("user\_id","product\_id").show(5)

2）对每个用户对product求和：

DataFrame转RDD处理：  
val rddRecords=op.rdd.map(x=>(x(0).toString,x(1).toString))  
.groupByKey()  
.mapValues(\_.toSet.mkString(","))

取出一条数据的结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 4][]

3）隐式转换：

3.1导包：import spark.implicits.\_

3.2代码：rddRecords.toDF("user\_id","product\_records")

结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 5][]

### 4.每个用户总商品数量以及去重后的商品数量 ###

总商品数量:  
orders.join(priors,"order\_id").groupBy("user\_id").count().show(5)

解法1：统计去重商品的数量：

![20190308141744371.png][]

结果：

![20190308141806389.png][]

解法2：

![20190308142220751.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 6][]

### 5.每个用户购买的平均每个订单的商品数量 ###

1）先求每个订单的商品数量【对订单做聚合count（）】  
val ordProCnt = priors.groupBy("order\_id").count()

2）求每个用户订单中商品个数的平均值【对user做聚合，avg（商品个数）】  
orders.join(ordProCnt,"order\_id").groupBy("user\_id").avg("count")  
(count出来的列名就叫count)

[20190308131318808.png]: /images/20220311/b17cf7c6bc5c475cbbfff55ff619d56b.png
[20190308131407410.png]: /images/20220311/5e8c6cfe1d7d47d5ba399cec40d7a0d4.png
[20190308131428643.png]: /images/20220311/564a3ccb81474568ba3cbd32736bc697.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70]: /images/20220311/abc4afe595b74d0f9fba7fa2e5945cd4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1]: /images/20220311/87cf4d57413c4416b23cd58a11113102.png
[20190308125437848.png]: /images/20220311/f0fb89ffce9544cc9c5bdb74f91625bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 2]: /images/20220311/aaaab1c3d4da437295334b5871f31b0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 3]: /images/20220311/d5ba05da657e419595e34967e4cd972f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 4]: /images/20220311/0bb8acc1d10a4e43a57bf3db06c3db81.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 5]: /images/20220311/7de18db7257342288b5c48f49f91a0c7.png
[20190308141744371.png]: /images/20220311/b30871942cba4d3d99599f089220ea34.png
[20190308141806389.png]: /images/20220311/95c95229a1ec4d4eba8a03febabace02.png
[20190308142220751.png]: /images/20220311/3f2c788bf5fc453f864baa32f29fd884.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 6]: /images/20220311/24bc6d4b845d4463a1f4d8a497c2c2de.png