Python-PySpark数据计算练习案例

import osfrom pyspark import SparkConf, SparkContext#####定义PySpark运行环境os.environ["PYSPARK_PYTHON"]="D:softpythonpython3104python.exe"#创建SparkConf类对象conf=SparkConf().setMaster("local[*]").setAppName("test_spark_app")#利用SparkConf构建SparkContext对象sContex=SparkContext(conf=conf)#打印pyspark的运行版本print(sContex.version)  #3.4.0#读取文件中的内容rdd_text=sContex.textFile("D:hello.txt")#将所有的单次都取出来并拼接成二元元组rdd_text_map=rdd_text.flatMap(lambda x: x.split(" ")).map(lambda x:(x,1))#利用reduceByKey进行分组聚合rdd_text_reduce=rdd_text_map.reduceByKey(lambda x,y:(x+y))#[('java', 7), ('kettle', 2), ('python', 6), ('kettlekettle', 1), ('spark', 4), ('pyspark', 3)]print(rdd_text_reduce.collect())
#停止SparkContext对象的运行sContex.stop()

"""完成练习案例：JSON商品统计需求：1. 各个城市销售额排名，从大到小2. 全部城市，有哪些商品类别在售卖3. 北京市有哪些商品类别在售卖"""import jsonimport osfrom pyspark import SparkConf, SparkContext#####定义PySpark运行环境os.environ["PYSPARK_PYTHON"]="D:softpythonpython3104python.exe"#创建SparkConf类对象conf=SparkConf().setMaster("local[*]").setAppName("test_spark_app")#利用SparkConf构建SparkContext对象sContex=SparkContext(conf=conf)#打印pyspark的运行版本print(sContex.version)  #3.4.0#读取文件中的内容rdd_orders=sContex.textFile("D:orders.txt")# TODO 需求1：城市销售额排名# 1.1 读取文件得到RDD# 1.2 取出一个个JSON字符串rdd_orders_json=rdd_orders.flatMap(lambda x: x.split("|"))#:['{"id":1,"timestamp":"2019-05-08T01:03.00Z","category":"平板电脑","areaName":"北京","money":"1450"}'.....print(f"rdd_orders_json:{rdd_orders_json.collect()}")# 1.3 将一个个JSON字符串转换为字典rdd_orders_dict=rdd_orders_json.map(lambda x:json.loads(x))#[{'id': 1, 'timestamp': '2019-05-08T01:03.00Z', 'category': '平板电脑', 'areaName': '北京', 'money': '1450'},print(f"rdd_orders_dict:{rdd_orders_dict.collect()}")# 1.4 取出城市和销售额数据# (城市，销售额)rdd_orders_map=rdd_orders_dict.map(lambda x:(x["areaName"],int(x["money"])))#[('北京', '1450'), ('北京', '1450'), ('北京', '8412'), ('上海', '1513').....print(f"rdd_orders_map:{rdd_orders_map.collect()}")# 1.5 按城市分组按销售额聚合res_01_rdd=rdd_orders_map.reduceByKey(lambda x,y:(x+y)).sortBy(lambda x:x[1],ascending=False,numPartitions=1)# 1.6 按销售额聚合结果进行排序#城市销售额排名:[('北京', 91556), ('杭州', 28831), ('天津', 12260), ('上海', 1513), ('郑州', 1120)]print(f"城市销售额排名:{res_01_rdd.collect()}")# TODO 需求2：全部城市有哪些商品类别在售卖# 2.1 取出全部的商品类别# 2.2 对全部商品类别进行去重res_02_rdd=rdd_orders_dict.map(lambda x:x["category"]).distinct()#全部城市有哪些商品类别在售卖:['平板电脑', '家电', '书籍', '手机', '电脑', '家具', '食品', '服饰']print(f"全部城市有哪些商品类别在售卖:{res_02_rdd.collect()}")# TODO 需求3：北京市有哪些商品类别在售卖# 3.1 过滤北京市的数据beijing_orders_rdd=rdd_orders_dict.filter(lambda x:x["areaName"]=='北京')# 3.2 取出全部商品类别# 3.3 进行商品类别去重res_03_rdd=beijing_orders_rdd.map(lambda x:x["category"]).distinct()#北京市有哪些商品类别在售卖:['平板电脑', '家电', '书籍', '手机', '电脑', '家具', '食品', '服饰']print(f"北京市有哪些商品类别在售卖:{res_03_rdd.collect()}")