秋刀鱼在做梦

这个屌丝很懒，什么也没留下！

热门标签

Python大数据处理库 PySpark实战总结二_大数据处理库pyspark实战 pdf下载

作者：秋刀鱼在做梦 | 2024-08-19 14:19:26

踩

大数据处理库pyspark实战 pdf下载

Python大数据处理库 PySpark实战二

- Pyspark建立Spark RDD

Pyspark建立Spark RDD

每个RDD可以分成多个分区，每个分区可以看作是一个数据集片段，可以保存到Spark集群中的不同节点上
RDD自身具有容错机制，且是一种只读的数据结构，只能通过转换生成新的RDD；一个RDD通过分区可以多台机器上并行处理；可将部分数据缓存在内存中，可多次重用；当内存不足时，可把数据落到磁盘上
创建RDD的方法
- parallelize(集合，分区数)
- range sc.range(1,10,2) 开始结束步长
- 使用HDFS建立RDD

pyspark shell

 #pyspark shell
 rdd = sc.parallelize(["hello world","hello spark"]);
 rdd2 = rdd.flatMap(lambda line:line.split(" "));
 rdd3 = rdd2.map(lambda word:(word,1));
 rdd5 = rdd3.reduceByKey(lambda a, b : a + b);
 rdd5.collect();
 quit();
1
2
3
4
5
6
7

VScode

 # vscode
 #pip install findspark
 #fix:ModuleNotFoundError: No module named 'pyspark'
 import findspark
 findspark.init()
 
 #############################
 from pyspark import SparkConf, SparkContext
 
 # 创建SparkContext
 conf = SparkConf().setAppName("WordCount").setMaster("local[*]")
 sc = SparkContext(conf=conf)
  
 rdd = sc.parallelize(["hello world","hello spark"]);
 rdd2 = rdd.flatMap(lambda line:line.split(" "));
 rdd3 = rdd2.map(lambda word:(word,1));
 rdd5 = rdd3.reduceByKey(lambda a, b : a + b);
 #print，否则无法显示结果
 #[('spark', 1), ('hello', 2), ('world', 1)]
 print(rdd5.collect());
 #防止多次创建SparkContexts
 sc.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Jupyter notebook

 #jupyter
 from pyspark.sql import SparkSession
 spark = SparkSession.builder.master("local[*]").appName("WordCount").getOrCreate();
 sc = spark.sparkContext
 rdd = sc.parallelize(["hello world","hello spark"]);
 rdd2 = rdd.flatMap(lambda line:line.split(" "));
 rdd3 = rdd2.map(lambda word:(word,1));
 rdd5 = rdd3.reduceByKey(lambda a, b : a + b);
 #print，否则无法显示结果
 #[('spark', 1), ('hello', 2), ('world', 1)]
 print(rdd5.collect());
 #防止多次创建SparkContexts
 sc.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13

动作算子

collect 把RDD类型数据转化为数组同时从集群中拉取数据dirver端

stats 返回RDD元素的计数、均值、方差、最大值和最小值

countByKey 统计RDD[K,V]中每个K的数量每个相同的K 结果加一不是把V的值相加

first：返回RDD中一个元素
max：返回最大的一个元素
sum：返回和
take：返回前n个元素
top：返回排序后的前n个元素降序 top(10,key=str):按照字典序排序前10个
count：返回个数
collect ：把RDD类型数据转化为数组同时从集群中拉取数据dirver端
collectAsMap：把键值RDD转换成Map映射保留其键值结构
countByKey：统计RDD[K,V]中每个K的数量每个相同的K 结果加一不是把V的值相加

countByValue ：统计一个RDD中各个Value出现的次数，返回字典，key是元素的值，value是出现的次数/

sc.parallelize(range(2,100)) 等价于 sc.range(2,100)

rdd3 = sc.parallelize([("a",1),("a",1),("b",2),("a",1)])
print(rdd3.countByKey())
#defaultdict(<class 'int'>, {'a': 3, 'b': 1})
print(rdd3.countByValue())
#defaultdict(<class 'int'>, {('a', 1): 3, ('b', 2): 1})
1
2
3
4
5
6
7

stats：返回RDD元素的计数、均值、方差、最大值和最小值

rdd = sc.parallelize(range(100))
print(rdd.stats())
#(count: 100, mean: 49.5, stdev: 28.86607004772212, max: 99, min: 0)
1
2
3

aggregate ： aggregate(zeroValue,seqOp,combOp) 使用seqOP函数和给定的zeroValue聚合每个分区上的元素，然后用CombOp和zeroValue聚合所有分区结果

data=[1,3,5,7,9,11,13,15,17]
rdd=sc.parallelize(data,2)
print(rdd.glom().collect()) 
# [[1, 3, 5, 7], [9, 11, 13, 15, 17]]
seqOp = (lambda x, y: (x[0] + y, x[1] + 1))  #求和 和 个数
combOp = (lambda x, y: (x[0] + y[0], x[1] + y[1])) 
a=rdd.aggregate((0,0),seqOp,combOp)
#(81, 9)=(0,0)+(16,4)+(65,5)
1
2
3
4
5
6
7
8

变换算子

coalesce 重新分区
filter 过滤
map 每个元素转换
flatmap 每个元素转换并扁平化
mapPartitions 按分区转换
mapValues KV格式保留k 对v操作
reduce 减少个数； reducebykey KV格式对v操作减少元素个数
join 内链接； fullOuterJoin 全外部连接
groupBy 函数的返回作为k 分组；groupByKey KV中的K分组
keys 、values获取对应序列
zip 元素相同一一对应
union 合并； substract 减法； intersection 交集； certesian交集
cache、persist 缓存
glom 查看分区状态
sortBy：对RDD元素进行排序

coalesce：rdd.coalesce(numPartitions,[isShuffle=False]) 将RDD进行重新分区，分区过程中是否进行混洗操作

rdd=sc.parallelize([1, 2, 3, 4, 5], 3).glom()
#[[1], [2, 3], [4, 5]]
rdd2 = sc.parallelize([1, 2, 3, 4, 5, 6], 3).coalesce(1,False)
#[1, 2, 3, 4, 5, 6]
1
2
3
4

repartition: 和coalesce(1,True) 一样重新分区并混洗
distinct ：去重

filter：返回满足过滤函数为True的元素构成 filter(lambda x: x%2 == 0)

#filter
rdd5 = sc.parallelize([1,2,3,4,5]).filter(lambda x: x%2 == 0)
print(rdd5.collect())
[2,4]
1
2
3
4

map：对RDD每个元素按照func定义的逻辑处理，在统计单词个数中常用rdd.map(func,preservesPartitioning=Flase)

rdd = sc.parallelize(["b", "a", "c", "d"])
rdd2 = rdd.map(lambda x: (x, 1))
#[('b', 1), ('a', 1), ('c', 1), ('d', 1)]
1
2
3

flatMap：对RDD中每一个元素按照func的处理逻辑操作，并将结果扁平化处理

#faltMap
rdd5 = sc.parallelize([1,2,3,4,5]).flatMap(lambda x:[(x,1)])
print(rdd5.collect())
[(1, 2), (2, 4), (3, 6), (4, 8), (5, 10)]
1
2
3
4

flatMapValues：对RDD元素格式为KV对中的Value进行func定义的逻辑处理，形成新的KV，并把结果扁平化处理

#flatMapValues
rdd = sc.parallelize([("a", [1, 2, 3]), ("c", ["w", "m"])])
ret = rdd.flatMapValues(lambda x: x)
#[('a', 1), ('a', 2), ('a', 3), ('c', 'w'), ('c', 'm')]
1
2
3
4

mapPartitions：RDD每个分区中元素按照定义的逻辑返回处理，并分别返回值

rdd = sc.parallelize([1, 2, 3, 4 , 5], 2)
def f(iter): 
    yield sum(iter) #yield的作用是把函数变成generator，返回的是iterable对象

rdd2 = rdd.mapPartitions(f)
print(rdd2.collect())
#[3,12]
1
2
3
4
5
6
7

mapValues：对KV格式的RDD中的每个元素应用函数，K值不变且保留原始分区, 对Value操作

rdd = sc.parallelize([("a", ["hello", "spark", "!"]), ("b", ["cumt"])])
rdd2 = rdd.mapValues(lambda x:len(x))
#[('a', 3), ('b', 1)]
1
2
3

mapPartitionsWithIndex：RDD每个分区中元素按照定义的逻辑返回处理，跟踪原始分区的索引

rdd = sc.parallelize([1, 2, 3, 4 ,5 ,6], 3)
def f(index, iter): 
  #分区索引 0,1,2
  print(index)
  for x in iter:
    #1,2;3,4;5,6
    print(x)
    yield index
ret = rdd.mapPartitionsWithIndex(f).sum()
#3=0+1+2
print(ret)	
1
2
3
4
5
6
7
8
9
10
11

reduce ：按照func对RDD元素计算，减少元素个数

rdd = sc.parallelize([1, 2, 3, 4, 5])
ret = rdd.reduce(lambda x,y : x+y)
15
1
2
3

reduceByKey : 对KV的数据进行运算，减少元素个数

rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 2),("b", 3)])
rdd2 = rdd.reduceByKey(lambda x,y:x+y)
#[('a', 3), ('b', 4)]
1
2
3

join：包含自身和另一个匹配键的所有成对元素，每对元素以(k,(v1,v2))元组返回，其中(k,v1)在自身，(k,v2)在另一个中

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("a", 3)])
ret = x.join(y).collect()
#[('a', (1, 2)), ('a', (1, 3))]
1
2
3
4

fullOuterJoin : 全外部连接没有匹配到就是None

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("c", 8)])
rdd = x.fullOuterJoin(y)
# [('a', (1, 2)), ('b', (4, None)), ('c', (None, 8))]
1
2
3
4

leftOuterJoin 和 rightOuterJoin : 左外连接和右外连接

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("c", 8)])
rdd = x.leftOuterJoin(y)
#[('b', (4, None)), ('a', (1, 2))]
rdd = x.rightOuterJoin(y)
#[('c', (None, 8)), ('a', (1, 2))]
1
2
3
4
5
6

groupBy ：groupBy(func,numPartitions=None,partitionFunc=<function portable_hash) 函数的返回作为key，通过key对其元素进行分组，返回新的RDD

rdd = sc.parallelize([1, 2, 3, 4, 5, 10])
rdd = rdd.groupBy(lambda x:x%2)
result = rdd.collect()
#[(0, <pyspark.resultiterable.ResultIterable object at 0x110ef9c50>), (1, <pyspark.resultiterable.ResultIterable object at 0x110ef94d0>)]
ret = sorted([(x, sorted(y)) for (x, y) in result])
#[(0, [2, 4, 10]), (1, [1, 3, 5])]
1
2
3
4
5
6

groupByKey : 将RDD中每个键的值分组为单个序列，用numsPartitions分区对生成的RDD进行哈希分区 如果求和或平均值建议使用reduceByKey 或 AggregateByKey

rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
rdd2 = rdd.groupByKey().mapValues(lambda x: sum(x))
rdd3 = rdd.reduceByKey(lambda x,y: x+y) #和rdd2一样
# [('a', 2), ('b', 1)]
print(sorted(rdd2.collect()))
1
2
3
4
5

keyBy：将原有RDD中的元素作为Key，Key通过func返回值作为value创建一个元组

rdd = sc.parallelize(range(0,3))
rdd = rdd.keyBy(lambda x: x*x)
#[(0, 0), (1, 1), (4, 2)]
1
2
3

keys：获取KV格式中的Key序列，返回新的RDD

rdd1 = sc.parallelize([("a",1),("b",2),("a",3)])
print(rdd1.keys().collect())
#['a', 'b', 'a']
1
2
3

values：获取KV格式中的Value序列，返回新的RDD

rdd1 = sc.parallelize([("a",1),("b",2),("a",3)])
print(rdd1.keys().collect())
#[1, 2, 3]
1
2
3

zip：rdd.zip(otherRDD)将第一个RDD中的元素作为Key，第二个RDD中的作为Value组成新的RDD，两个RDD的元素个数相同

x = sc.parallelize(range(1,6))
y = sc.parallelize(range(801, 806))
print(x.zip(y).collect())
#[(1, 801), (2, 802), (3, 803), (4, 804), (5, 805)]
#x,y长度必须相等
1
2
3
4
5

zipWithIndex：RDD元素作为key，索引作为Value

rdd = sc.parallelize(["a", "b", "c", "d"], 3)
print(rdd.zipWithIndex().collect())
#[('a', 0), ('b', 1), ('c', 2), ('d', 3)]
1
2
3

union：第一个RDD元素和第二个的合并

dd =sc.parallelize(range(1,10))
rdd2 =sc.parallelize(range(11,20))
rdd3 = rdd.union(rdd2)
#[1, 2, 3, 4, 5, 6, 7, 8, 9, 11, 12, 13, 14, 15, 16, 17, 18, 19]
1
2
3
4

subtract：第一个中排出第二个中的元素

x = sc.parallelize([("a", 1), ("b", 4), ("b", 5), ("a", 3)])
y = sc.parallelize([("a", 1), ("b", 5)])
z = x.subtract(y)
#[('b', 4), ('a', 3)]
1
2
3
4

subtractByKey ：从元素为KV格式的RDD中除掉另一个，只要Key一样就删除

x = sc.parallelize([("a", 1), ("b", 4), ("c", 5), ("a", 3)])
y = sc.parallelize([("a", 7), ("b", 0)])
z = x.subtractByKey(y)
#[('c', 5)]
1
2
3
4

intersection：返回交集并去重

rdd1 = sc.parallelize([("a", 2), ("b", 1), ("a", 2),("b", 3)])
rdd2 = sc.parallelize([("a", 2), ("b", 1), ("e", 5)])
ret = rdd1.intersection(rdd2).collect()
#('a', 2), ('b', 1)]
1
2
3
4

certesian：返回两个RDD的笛卡尔积元素较多可能出现内存不足情况

rdd = sc.parallelize([1, 2])
rdd2 = sc.parallelize([3, 7])
rdd3 = sorted(rdd.cartesian(rdd2).collect())
#[(1, 3), (1, 7), (2, 3), (2, 7)]
print(rdd3)
1
2
3
4
5

sortBy：对RDD元素进行排序，sortBy(keyfuc,ascending=True,numPartitions=None)，默认升序

rdd = [('a', 6), ('f', 11), ('c', 7), ('d', 4), ('e', 5)]
rdd2 = sc.parallelize(rdd).sortBy(lambda x: x[0])
#[('a', 6), ('c', 7), ('d', 4), ('e', 5), ('f', 2)]
rdd3 = sc.parallelize(rdd).sortBy(lambda x: x[1])
#[('f', 2), ('d', 4), ('e', 5), ('a', 6), ('c', 7)]
rdd3 = sc.parallelize(rdd).sortBy(lambda x: x[1],False)
#[('c', 7), ('a', 6), ('e', 5), ('d', 4), ('f', 2)]
1
2
3
4
5
6
7

sortByKey : 按照Key排序 sortByKey(ascending=True,numPartitions=None,keyfunc=)

x = [('a', 6), ('f', 2), ('c', 7), ('d', 4), ('e', 5)]
rdd = sc.parallelize(x).sortByKey(True, 1)
#[('a', 6), ('c', 7), ('d', 4), ('e', 5), ('f', 2)]
print(rdd.collect())
1
2
3
4

takeOrdered：RDD中获取排序后的前num个元素构成RDD，默认升序，可支持可选函数

rdd =sc.parallelize(range(2,100))
print(rdd.takeOrdered(10))
#[2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
print(rdd.takeOrdered(10, key=lambda x: -x))
#[99, 98, 97, 96, 95, 94, 93, 92, 91, 90]
1
2
3
4
5

takeSample：takeSample(withReplacement,num,seed=None) 抽样出固定大小的子数据集合，第一个参数布尔值表示是否可以多次抽样，第二个抽样的个数，第三个随机数生成器种子

dd =sc.parallelize(range(2,10))
print(rdd.takeSample(True, 20, 1))
#True代表一个元素可以出现多次
#[5, 9, 5, 3, 2, 2, 7, 7, 5, 7, 9, 9, 5, 3, 2, 4, 5, 5, 6, 8]
print(rdd.takeSample(False, 20, 1))
#False代表一个元素只能出现1次
#[5, 8, 3, 7, 9, 2, 6, 4]
1
2
3
4
5
6
7

sample : sample(withReplacement,fraction,seed) 第二个参数抽样比例[0,1]

rdd = sc.parallelize(range(100), 1)
ret = rdd.sample(False, 2, 1)
#可能输出[9, 11, 13, 39, 49, 55, 61, 65, 90, 91, 93, 94]
1
2
3

randomSplit：按照权重对RDD随机切分，返回多个RDD构成的列表

rdd = sc.parallelize(range(100), 1)
rdd1, rdd2 = rdd.randomSplit([2, 3], 10)
print(len(rdd1.collect())) #40
print(len(rdd2.collect())) #60
1
2
3
4

loopup：根据key值从RDD中找到相关的元素，返回KV中的V

rdd = sc.parallelize([('a', 'b'), ('c', 'd')])
print(rdd.lookup('a')) #['b']
1
2

fold：对RDD每个元素按照func的逻辑进行处理fold(value,func) func有两个参数a,b a的初始值为value，后续为累加值，b代表当前元素值 可以用来累加累乘
```
#fold
ret=sc.parallelize([1, 2, 3, 4, 5]).fold(0, lambda x,y:x+y)
#15
ret=sc.parallelize([1, 2, 3, 4, 5]).fold(1, lambda x,y:x*y)
#120
1
2
3
4
5
```

foldByKey：对RDD元素格式为KV对中的Key进行func定义的逻辑处理，可以用来分组累加累乘

#foldByKey
rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3),("b", 5)])
rdd2=rdd.foldByKey(0, lambda x,y:x+y)
# [('a', 4), ('b', 7)]
rdd3=rdd.foldByKey(1, lambda x,y:x*y)
# [('a', 3), ('b', 10)]
1
2
3
4
5
6

foreach：对RDD每个元素按照func定义的逻辑处理

foreachPartion：对RDD每个分区中的元素按照func定义逻辑处理，一般来说foreachPartion效率比foreach高，是一次性处理一个partition数据，在写数据库的时候，性能比map高很多

rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3),("b", 5)])
def f(x):
    print(x)
    return (x[0],x[1]*2)
    
def f2(iter):
    for x in iter:
        print(x)
        
ret = rdd.foreach(f)
ret2 = sc.parallelize([1,2,3,4,5,6,7,8],2).foreachPartition(f2)
1
2
3
4
5
6
7
8
9
10
11

aggregateByKey：aggregate(zeroValue,seqFunc,combFunc,numPartitions=None,partitionFunc=) 使用seqFunc函数和给定的zeroValue聚合每个分区上的元素，然后用CombFunc和zeroValue聚合所有分区结果

data=[("a",1),("b",2),("a",3),("b",4),("a",5),("b",6),("a",7),("b",8),("a",9),("b",10)]
rdd=sc.parallelize(data,2)
print(rdd.glom().collect())
#[[('a', 1), ('b', 2), ('a', 3), ('b', 4), ('a', 5)], [('b', 6), ('a', 7), ('b', 8), ('a', 9), ('b', 10)]]
def seqFunc(x,y):
	return x + y
def combFunc(x,y):
	return x + y
a=rdd.aggregateByKey(0,seqFunc,combFunc)
# [('b', 30), ('a', 25)]
print(a.collect())
1
2
3
4
5
6
7
8
9
10
11

combineByKey:

createCombiner： V => C 这个函数把当前的值作为参数可以对其做一些操作并返回
mergeValue ：(C,V) => C 把元素V合并到之前的元素C上（这个操作在每个分区内进行）
mergeCombiners：(C,C) => C 把2个元素合并（这个操作在不同分区间进行）

a = [1,2]
b = [10,11]
a.extend(b) #[1, 2, 10, 11]
a.append(b) #[1, 2, [10, 11]]

#combineByKey
rdd = sc.parallelize([("a", 1), ("b", 3), ("a", 2),("b", 4)],2)
def to_list(a):
    return [a]
def append(a, b): #分区合并
    a.append(b)
    return a
def extend(a, b):#不同分区合并
    a.extend(b)
    return a
print(rdd.glom().collect())
ret = sorted(rdd.combineByKey(to_list, append, extend).collect())
#[[('a', 1), ('b', 3)], [('a', 2), ('b', 4)]]
#[('a', [1, 2]), ('b', [3, 4])]  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

glom：把RDD中每一个分区的元素T转换成Array[T]，每个分区只有一个数组元素

#glom
rdd2 = sc.parallelize([1,2,3,4,5],3)
print(rdd2.collect())
#[1, 2, 3, 4, 5]
print(rdd2.glom().collect())
#[[1], [2, 3], [4, 5]]
print(rdd2.coalesce(1).glom().collect())
#[[1, 2, 3, 4, 5]]
1
2
3
4
5
6
7
8

cache ：缓存默认存储级别(MEMORY_ONLY)
persist : 缓存可以定制存储级别 storageLevel
saveAsTextFile：保存RDD文件作为一个对象，

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/秋刀鱼在做梦/article/detail/1002661

Python大数据处理库 PySpark实战 总结二_大数据处理库pyspark实战 pdf下载