数据输入

PySpark支持多种数据的输入, 在输入完成后, 都会得到一个: RDD类的对象
RDD全称为: 弹性分布式数据集 (Resilient Distributed Datasets)
PySpark针对数据的处理, 都是以RDD对象作为载体, 即:
- 数据存储在RDD内
- 各类数据的计算方法, 也都是RDD的成员方法
- RDD的数据计算方法, 返回值依旧是RDD对象

Python数据容器转RDD对象

PySpark支持通过SparkContext对象的parallelize成员方法将容器转换成PySpark的RDD对象

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

#通过parallelize方法将Python对象加载到Spark内，成为RDD对象
rdd1 = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = sc.parallelize((1, 2, 3, 4, 5))
rdd3 = sc.parallelize("abcdefg")
rdd4 = sc.parallelize({1, 2, 3, 4, 5})
rdd5 = sc.parallelize({"key1":"value1", "key2":"value2"})

#查看RDD里面的内容时，需要用collect()方法
#注意：字符串会被拆分成一个个的字符存入RDD对象
#     字典仅有key会被存入RDD对象
print(rdd1.collect())#[1, 2, 3, 4, 5]
print(rdd2.collect())#[1, 2, 3, 4, 5]
print(rdd3.collect())#['a', 'b', 'c', 'd', 'e', 'f', 'g']
print(rdd4.collect())#[1, 2, 3, 4, 5]
print(rdd5.collect())#['key1', 'key2']

sc.stop()

读取文件转RDD对象

PySpark支持通过SparkContext入口对象读取文件，来构建出RDD对象

文件内容:

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.textFile("E:/hello.txt")
print(rdd.collect())#['张三', '李四', 'wangwu']

sc.stop()

数据计算

PySpark的数据计算,都是基于RDD对象来进行的,RDD对象内置丰富的: 成员方法(算子)

map算子

将RDD的数据一条条处理并返回新的RDD

对于返回值是RDD的算子，可以使用链式调用的方式多次调用算子，如：map().map().map(). ...

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.parallelize([1,2,3,4,5])

def func(data):
    return data * 10

rdd2 = rdd.map(func)
print(rdd2.collect())# [10, 20, 30, 40, 50]

sc.stop()

flatMap算子

对rdd执行map操作，然后进行解除嵌套操作

# 嵌套的list
lst = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

# 如果解除了嵌套
lst = [1, 2, 3, 4, 5, 6, 7, 8, 9]

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.parallelize(["a,b,c","b,c,d","c,d,e"])

rdd2 = rdd.flatMap(lambda x: x.split(","))
print(rdd2.collect())# ['a', 'b', 'c', 'b', 'c', 'd', 'c', 'd', 'e']

sc.stop()

reduceByKey算子

针对KV型RDD，自动按照key分组，然后根据提供的聚合逻辑，完成组内数据(value)的聚合操作

KV型RDD：二元元祖（元组内部只有两个元素，第一个元素是key，第二个元素是value）

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

# 准备一个RDD
rdd = sc.parallelize([('男', 99), ('男', 88), ('女', 99), ('女', 66)])
# 求男生和女生两个组的成绩之和
rdd2 = rdd.reduceByKey(lambda a, b: a + b)
print(rdd2.collect())# [('男', 187), ('女', 165)]

sc.stop()

练习1：

统计文件中单词的数量，文件内容如下：

itheima itheima itcast itheima spark python spark python itheima itheima itcast itcast itheima python python python spark pyspark pyspark itheima python pyspark itcast spark

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.textFile("E:/hello.txt")

rdd2 = rdd.flatMap(lambda line: line.split(" "))
rdd3 = rdd2.map(lambda x: (x, 1))
rdd4 = rdd3.reduceByKey(lambda a,b: a+b)

print(rdd4.collect())
sc.stop()

filter算子

过滤想要保留的数据，返回值必须是true(数据被保留)或false(数据被丢弃)

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.parallelize([1,2,3,4,5])

rdd2 = rdd.filter(lambda x: x % 2 == 1)
#为True时保留结果，所以会保留所有的奇数

print(rdd2.collect())#[1, 3, 5]
sc.stop()

distinct算子

对RDD数据进行去重，返回新的RDD。语法：rdd.distinct() 无需传参

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.parallelize([1,1,2,2,4,3,3,4,4,4,3,2,2])

rdd2 = rdd.distinct()

print(rdd2.collect())#[1, 2, 3, 4]
sc.stop()

sortBy算子

对RDD数据进行排序，基于指定的排序依据

rdd.sortBy(func, ascending=False, numPartitions=1)
func: (T) → U: 告知按照rdd中的哪个数据进行排序，比如 lambda x: x[1] 表示按照rdd中的第二列元素进行排序
ascending True升序 False 降序
numPartitions: 用多少分区排序

将练习1的结果按照单词数量从大到小的顺序输出：

from pyspark import SparkConf, SparkContext

#需要手动写出python解释器的位置才能正常运行
import os
os.environ['PYSPARK_PYTHON'] = "C:\DevApp\python3.10.4\python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

rdd = sc.textFile("E:/hello.txt")

rdd2 = rdd.flatMap(lambda line: line.split(" "))
rdd3 = rdd2.map(lambda x: (x, 1))
rdd4 = rdd3.reduceByKey(lambda a,b: a+b)
rdd5 = rdd4.sortBy(lambda x:x[1],ascending=False,numPartitions=1)

print(rdd5.collect())
#[('itheima', 7), ('python', 6), ('itcast', 4), ('spark', 4), ('pyspark', 3)]
sc.stop()