1226 练习

1226 练习 1 、数据背景 该数据每日进行采集汇总。数据范围涵盖全国主要省份（港澳台、西藏、海南暂无数 据）的 180+ 的大型农产品批发市场， 380+ 的农产品品类（由于季节性和地域性 等特点，每日的数据中不一定会涵盖全部的农产品品类）。 2 、数据类型 表 1 农产品批发数据数据字典 表 2 省份名称与对应英文缩写数据字典 四、功能要求 1. 数据准备（ 15 分） a) 请在 HDFS 中创建目录 /app/data/allprovinces ，并将 allprovinces.txt, products.txt 传 到该目录。 b) 请 在 HDFS 中 创 建 目 录 /app/data/events/products ， 并 将 products.txt 从 3 /app/data/events/allprovinces 转到该目录，后面均从 HDFS 上加载对应数据。 2. 创建 Hbase 数据表（ 10 分） 在 Hbase 中创建数据库表 province_market ，该表 RowKey 为省份首字母缩写，并有两 个列族 market 和 info ， market 用于保存该省份的市场总数（ maket_count ）， info 用于 保存该省的全称（ province_name ）。 3. 请在 Hive 中创建数据库 market ，并在 market 数据库中创建三个外表。 a) 外表 ex_allprovinces 指向 allprovinces.txt 。 b) 外表 ex_products 指向 products.txt 。 c) 外表 ex_province_market 指向 Hbase 中的 province_maket 表 。 d) 统计每个省份的农产品市场总数，并将每个省份的市场数量保存到 Hbase 的 province_market 表中（如该省无农产品市场，则数量为 0 ）。其中 RowKey 使用省 份名称的英文缩写（来自于 allprovinces.txt ），市场数量保存在 market 列族下 的 count 列中，省份名称保存在 info 列族下的 name 列中。 (30 分 ) 4. 在 Spark-Shell 中，将 products.txt 装载到 RDD 中，使用 RDD 对农产品种类统 计，并将结果输出到控制台上。（ 20 分） ①根据农产品类型数量，统计排名前 3 名的省份。 ②根据农产品类型数量，统计每个省份排名前 3 名的农产品市场。 5. 在 Spark-Shell 中，将 products.txt 装载成 Dataframe ，计算山西省每种农产 品的价格波动趋势，即计算每天价格均值，并将结果输出到控制台上。（ 25 分） 某种农产品的价格均值计算公式： PAVG = (PM1+PM2+...+PMn-max(P)-min(P))/(N-2) 其中， P 表示价格， Mn 表示 market ，即农产品市场。 PM1 表示 M1 农产品市场的该 产品价格， max(P) 表示价格最大值， min(P) 价格最小值。

第一题​​​​​​​上传

hdfs dfs -mkdir /app/data/allprovinces

hdfs dfs -put /opt/1226/products.txt /app/data/allprovinces/

 hdfs dfs -put /opt/1226/allprovinces.txt /app/data/allprovinces/

 hdfs dfs -mv  /app/data/allprovinces/products.txt  /app/data/events/products

1. 创建 Hbase 数据表

1. 请在 Hive 中创建数据库 market，并在 market 数据库中创建三个外表

 hive --service hiveserver2

hive --service metastore

create database market;

create database market;
use market;
create external table ex_allprovinces(
    name string,
    abbr string
)
row format delimited fields terminated by 't'
location '/app/data/allprovinces/';

create external table ex_products(
    name string,
    price float,
    craw_time string,
    market string,
    province string,
    city string
)
row format delimited fields terminated by 't'
location '/app/data/events/products';

3c

create external table ex_province_market(
   rowkey string,
   marketCount string,
   provicneName string
)
stored by 'org.apache.hadoop.hive.hbase.HbaseStorageHandler'
with serdeproperties
("hbase.columns.mapping" = ":key,market:count,info:name")
tblproperties ("hbase.table.name" = "exam:province_market");

4d插入那个hbase表的语句

with t1
    as (
select province,count(distinct(market))con from ex_products where province!="" group by province )
insert into table ex_province_market
select t2.abbr,t1.con,t1.province from t1 inner join ex_allprovinces t2 on t1.province=t2.name ;


select * from ex_province_market;

4.在 Spark-Shell 中，将 products.txt 装载到 RDD 中，使用 RDD 对农产品种类统

计，并将结果输出到控制台上。

//rdd 1题
val rdd = sc.textFile("hdfs://192.168.10.136:9000/app/data/events/products/products.txt")
rdd.map(_.split("t")).filter(_.size==6).map(x=>{
  (x(4),x(0))
}).groupByKey().map(x=>(x._1,x._2.toArray.distinct.size))
  .repartition(1).sortBy(_._2,false).take(3).foreach(println)

//rdd 2题
val lines = sc.textFile("hdfs://192.168.10.136:9000/app/data/events/products/products.txt")
val tmp=lines.distinct.filter(_.split("t").length==6)

tmp.map(line=>{
  val fields: Array[String] = line.split("t")
  ((fields(4),fields(3)),1)
}).reduceByKey(_+_).map({case((province,market),nums)=>(province,(market,nums))})
  .groupByKey().mapValues(x=>x.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)).foreach(println)

5.在 Spark-Shell 中，将 products.txt 装载成 Dataframe，计算山西省每种农产

品的价格波动趋势，即计算每天价格均值，并将结果输出到控制台上。

val df: Dataframe = sc.textFile("hdfs://192.168.10.136:9000/app/data/events/products/products.txt")
  .map(_.split("t")).filter(_.length == 6).map(x => {
  (x(0), x(1), x(2), x(3), x(4), x(5))
}).toDF("name", "price", " craw_time", "market", "province", "city")
 

df.where("province='山西'").groupBy("name").agg(
  sum($"price").as("sum"),
  min($"price").as("min"),
  count($"price").as("count"),
  max($"price").as("max")
).withColumn("pavg",
  when($"count">2, ($"sum" - $"max" - $"min") / ($"count" - 2)).otherwise($"sum"/$"count"))
  .show()

这里空值是因为有的种类只有一个价格 减掉2变成负数所以最后得到null 题目没标注 就不特意处理了 保持原汁原味