（PySpark）RDD实验实战——求商品销量排行

 实验环境：

提前准备好 findspark，pyspark，py4j 等库

import findspark
from pyspark import SparkContext, SparkConf

findspark.init()

# 初始化 spark，默认为你所设定的环境变量 

conf = SparkConf().setAppName("jsytest").setMaster("local[4]")

# 创建一个 SparkConf 对象，用于配置 Spark 应用程序，用 setAppName 来设置程序名称，#用 setMaster 来设置运行模式和线程数，这里为本地模式，4 个线程 

sc = SparkContext(conf=conf)

# 创建一个 SparkContext 对象，它是与 Spark 集群通信的主要接口
# sc.stop()  #关闭 spark 上下文 

goods = [("Book",30,50),("Pen",5,80),("Notebook",15,60),("Pencil",2,70),("Eraser",3,50)]

# 创建所需的数据集（商品名，价格，销量）

n=2

# 所需的前 n 个排序 

rdd = sc.parallelize(goods)

# 用 parallelize 方法将 goods 中的数据结构并行化成 RDD

rdd.sortBy(lambda x:x[2], ascending=True,numPartitions=3).collect()

#rdd.sortBy() 用于对 RDD 中的元素按照指定的排序键进行排序
#rdd.sortBy(keyfunc, ascending=True, numPartitions=None)
#keyfunc, 是从 RDD 的每个元素中提取用于排序的键，多分区的话可以通过指定 key 的排序，来达到操作目的
#ascending 表示排序的顺序。True 为升序，False 为降序。#numPartitions 表示最终返回结果 RDD 的分区数。#这里取的是数据中的第三分区销量作为 key，返回的值也是三个分区 

ss=rdd.sortBy(lambda x:x[2], ascending=True,numPartitions=3).collect()

## 把最终排序导入 ss 数组中 

c=rdd.count()-1

# 取数组上限 

ysj=0

# 记录循环次数 

while ysj

所有代码如下

import findspark
from pyspark import SparkContext, SparkConf
findspark.init()
conf = SparkConf().setAppName("jsytest").setMaster("local[4]")
sc = SparkContext(conf=conf
goods = [("Book",30,50),("Pen",5,80),("Notebook",15,60),("Pencil",2,70),("Eraser",3,50)]
n=2
rdd = sc.parallelize(goods)
rdd.sortBy(lambda x:x[2], ascending=True,numPartitions=3).collect()
ss=rdd.sortBy(lambda x:x[2], ascending=True,numPartitions=3).collect()
c=rdd.count()-1
ysj=0
while ysj

结果演示

 销售第 1 多的: ('Pen', 5, 80)
销售第 2 多的: ('Pencil', 2, 70)