spark案例解析（全国农产品市场与省份）

需求

(一) 数据描述
1、数据参数
该数据每日进行采集汇总。数据范围涵盖全国主要省份&＃xff08;港澳台西藏海南暂无数
据&＃xff09;的 180&＃43;的大型农产品批发市场&＃xff0c;380&＃43;的农产品品类&＃xff08;由于季节性和地域性
等特点&＃xff0c;每日的数据中不一定会涵盖全部的农产品品类&＃xff09;。
2、数据类型

&＃xff08;二&＃xff09;功能需求
1、农产品市场个数统计

1. 统计每个省份的农产品市场总数
2. 统计没有农产品市场的省份有哪些

2、农产品种类统计

1. 根据农产品类型数量&＃xff0c;统计排名前 3 名的省份
2. 根据农产品类型数量&＃xff0c;统计每个省份排名前 3 名的农产品市场

3、价格区间统计,

1. 计算山西省的每种农产品的价格波动趋势&＃xff0c;即计算每天价格均值。
2. 某种农产品的价格均值计算公式&＃xff1a;
   PAVG &＃61; (PM1&＃43;PM2&＃43;…&＃43;PMn-max§-min§)/(N-2)
   其中&＃xff0c;P 表示价格&＃xff0c;Mn 表示 market&＃xff0c;即农产品市场。PM1 表示 M1 农产品
   市场的该产品价格&＃xff0c;max§表示价格最大值&＃xff0c;min§价格最小值。

数据展示

allprovince.txt

河北
山西
辽宁
吉林
黑龙江
江苏
浙江
安徽
福建
江西
山东
河南
...


product.txt

生菜 2.00 2018/1/1 山西汾阳市晋阳农副产品批发市场 山西 汾阳
芹菜 2.40 2018/1/1 山西汾阳市晋阳农副产品批发市场 山西 汾阳
菜花 3.80 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
生姜 10.00 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
山药 8.00 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
芋头 5.50 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
小葱 1.50 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
...


用RDD算子解法&＃xff1a;

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit/*** Description :* CreateTime : 2019/10/916:41** &＃64;author TuYouXian* &＃64;since JDK1.8*/
class Test {val sc &＃61; new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))val source: RDD[String] &＃61; sc.textFile("data/product.txt")/*** 统计每个省份农产品批次市场个数* 已有数据&＃xff1a;省份 市场名称* 目标数据&＃xff1a;省份 市场数量*/&＃64;junit.Testdef product1(): Unit &＃61;{//1、读取数据//2、去重、过滤、列裁剪val result: collection.Map[String, Long] &＃61; source.filter(_.split("\t").size >&＃61; 5).map(line &＃61;>{val arr: Array[String] &＃61; line.split("\t")val province &＃61; arr(4)val product &＃61; arr(3)province-> product}).distinct()//3、数据处理//.map(item &＃61;> item._1 -> 1).reduceByKey(_&＃43;_).collect()//.groupByKey().map(item &＃61;> item._1->item._2.size).countByKey()//4、结果展示println(result)}/*** 统计没有农产品的省份有哪些* 已知数据&＃xff1a;有农产品省份* 目标数据&＃xff1a;无农产品省份*/&＃64;junit.Testdef product2(): Unit &＃61;{val allprovinces &＃61; sc.textFile("data/allprovince.txt")val provinces &＃61; source.filter(_.split("\t").size >&＃61; 5).map(line &＃61;> {val arr &＃61; line.split("\t")val province &＃61; arr(4)province}).distinct()val noprovinces: Array[String] &＃61; allprovinces.subtract(provinces).collect()println(noprovinces.toBuffer)}/*** 根据农产品的类型数量&＃xff0c;统计排名前三的省份* 已有数据&＃xff1a;农产品类型&＃xff0c;省份* 中间数据&＃xff1a;农产品类型数量&＃xff0c;省份* 目标数据&＃xff1a;排名前三省份*/&＃64;junit.Testdef product3(): Unit &＃61;{val result: Array[(String, Int)] &＃61; source.filter(_.split("\t").size >&＃61; 5).map(line &＃61;> {val arr: Array[String] &＃61; line.split("\t")val province &＃61; arr(4)val productType &＃61; arr(0)province -> productType}).distinct().groupByKey().map(item &＃61;> item._1 -> item._2.size).sortBy(_._2, false).take(3)println(result.toBuffer)}/*** 根据农产品类型数量&＃xff0c;统计每个省份排名前三的批发市场* 已有数据&＃xff1a;农产品类型&＃xff0c;省份&＃xff0c;批发市场* 中间数据&＃xff1a;&＃xff08;省份&＃xff0c;&＃xff08;&＃xff08;省份&＃xff0c;批发市场&＃xff09;&＃xff0c;农产品类型数量&＃xff09;&＃xff09;* 目标数据&＃xff1a;&＃xff08;省份&＃xff0c;批发市场&＃xff0c;农产品类型数量&＃xff09;前三数据* 也就是要根据省份分区&＃xff0c;求每个分区的前三*/&＃64;junit.Testdef product4(): Unit &＃61;{source.filter(_.split("\t").size >&＃61; 5).map(line &＃61;>{val arr &＃61; line.split("\t")val province &＃61; arr(4)val product &＃61; arr(3)val productType &＃61; arr(0)((province,product),productType)}).distinct().map(item &＃61;> item._1 -> 1).reduceByKey(_&＃43;_).groupBy(_._1._1)//&＃xff08;省份&＃xff0c;[&＃xff08;&＃xff08;省份&＃xff0c;批发市场&＃xff09;&＃xff0c;农产品类型数量&＃xff09;,&＃xff08;&＃xff08;省份&＃xff0c;批发市场&＃xff09;&＃xff0c;农产品类型数量&＃xff09;]&＃xff09;.flatMap(item &＃61;>{//需要压扁操作val province &＃61; item._1//val arr &＃61; item._2.toArrayval arr &＃61; item._2.toBufferarr.sortBy(_._2).reverse.take(3).map(item &＃61;> (province,item._1._2,item._2))}).foreach(println(_))}/*** 计算山西省的每种农产品的价格波动趋势&＃xff0c;即计算每天价格均值* PAVG &＃61; (PM1&＃43;PM2&＃43;...&＃43;PMn-max(P)-min(P))/(N-2)* 其中&＃xff0c;P 表示价格&＃xff0c;Mn 表示 market&＃xff0c;即农产品市场。PM1 表示 M1 农产品* 市场的该产品价格&＃xff0c;max(P)表示价格最大值&＃xff0c;min(P)价格最小值。* 已知数据&＃xff1a;农产品类型&＃xff0c;批发价格* 目标数据&＃xff1a;农产品类型&＃xff0c;价格均值* 注意有部分农产品的数量并不能满足上面公式可能会出NAN&＃xff0c;无穷数等结果*/&＃64;junit.Testdef product5(): Unit &＃61;{//1、过滤 去重 列裁剪source.filter(item&＃61;> item.split("\t").size>&＃61;5 && item.split("\t")(4)&＃61;&＃61;"山西").map(item&＃61;>{val arr &＃61; item.split("\t")val name &＃61; arr(0)val price &＃61; arr(1).toDouble//避免后续出现1/3 &＃61; 0的情况(name,price)})//2、分组//(面粉,CompactBuffer(3.44, 2.9, 3.44))//(大葱,CompactBuffer(2.8, 2.6, 2.4, 3.0, 3.1, 2.8, 3.0, 2.5)).groupByKey()//.foreach(println(_)).map(item&＃61;>{val data: Iterable[Double] &＃61; item._2val maxprice &＃61; data.maxval minprice &＃61; data.minval sumPrice &＃61; data.sumval size &＃61; data.sizeval avgPrice &＃61; (sumPrice-maxprice-minprice)/(size-2)(item._1,avgPrice)})//3、计算平均价格.foreach(println(_))//4、结果}
}

SparkSql解法&＃xff1a;