公眾號(hào)后臺(tái)回復(fù)關(guān)鍵字：pyspark，獲取本項(xiàng)目github地址。

本節(jié)將介紹RDD數(shù)據(jù)結(jié)構(gòu)的常用函數(shù)。包括如下內(nèi)容:

創(chuàng)建RDD

常用Action操作

常用Transformation操作

常用PairRDD的轉(zhuǎn)換操作

緩存操作

共享變量

分區(qū)操作

這些函數(shù)中，我最常用的是如下15個(gè)函數(shù)，需要認(rèn)真掌握其用法。

map

flatMap

mapPartitions

filter

count

reduce

take

saveAsTextFile

collect

join

union

persist

repartition

reduceByKey

aggregateByKey

import?findspark

#指定spark_home為剛才的解壓路徑,指定python路徑

spark_home?=?"/Users/liangyun/ProgramFiles/spark-3.0.1-bin-hadoop3.2"

python_path?=?"/Users/liangyun/anaconda3/bin/python"

findspark.init(spark_home,python_path)

import?pyspark

from?pyspark?import?SparkContext,?SparkConf

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

sc?=?SparkContext(conf=conf)

print(pyspark.__version__)

3.0.1

一，創(chuàng)建RDD

創(chuàng)建RDD主要有兩種方式，一個(gè)是textFile加載本地或者集群文件系統(tǒng)中的數(shù)據(jù)，

第二個(gè)是用parallelize方法將Driver中的數(shù)據(jù)結(jié)構(gòu)并行化成RDD。

#從本地文件系統(tǒng)中加載數(shù)據(jù)

file?=?"./data/hello.txt"

rdd?=?sc.textFile(file,3)

rdd.collect()

['hello?world',

'hello?spark',

'spark?love?jupyter',

'spark?love?pandas',

'spark?love?sql']

#從集群文件系統(tǒng)中加載數(shù)據(jù)

#file?=?"hdfs://localhost:9000/user/hadoop/data.txt"

#也可以省去hdfs://localhost:9000

#rdd?=?sc.textFile(file,3)

#parallelize將Driver中的數(shù)據(jù)結(jié)構(gòu)生成RDD,第二個(gè)參數(shù)指定分區(qū)數(shù)

rdd?=?sc.parallelize(range(1,11),2)

rdd.collect()

[1,?2,?3,?4,?5,?6,?7,?8,?9,?10]

二，常用Action操作

Action操作將觸發(fā)基于RDD依賴關(guān)系的計(jì)算。

collect

rdd?=?sc.parallelize(range(10),5)

#collect操作將數(shù)據(jù)匯集到Driver,數(shù)據(jù)過(guò)大時(shí)有超內(nèi)存風(fēng)險(xiǎn)

all_data?=?rdd.collect()

all_data

[0,?1,?2,?3,?4,?5,?6,?7,?8,?9]

take

#take操作將前若干個(gè)數(shù)據(jù)匯集到Driver，相比collect安全

rdd?=?sc.parallelize(range(10),5)

part_data?=?rdd.take(4)

part_data

[0,?1,?2,?3]

takeSample

#takeSample可以隨機(jī)取若干個(gè)到Driver,第一個(gè)參數(shù)設(shè)置是否放回抽樣

rdd?=?sc.parallelize(range(10),5)

sample_data?=?rdd.takeSample(False,10,0)

sample_data

[7,?8,?1,?5,?3,?4,?2,?0,?9,?6]

first

#first取第一個(gè)數(shù)據(jù)

rdd?=?sc.parallelize(range(10),5)

first_data?=?rdd.first()

print(first_data)

0

count

#count查看RDD元素?cái)?shù)量

rdd?=?sc.parallelize(range(10),5)

data_count?=?rdd.count()

print(data_count)

10

reduce

#reduce利用二元函數(shù)對(duì)數(shù)據(jù)進(jìn)行規(guī)約

rdd?=?sc.parallelize(range(10),5)

rdd.reduce(lambda?x,y:x+y)

45

foreach

#foreach對(duì)每一個(gè)元素執(zhí)行某種操作，不生成新的RDD

#累加器用法詳見共享變量

rdd?=?sc.parallelize(range(10),5)

accum?=?sc.accumulator(0)

rdd.foreach(lambda?x:accum.add(x))

print(accum.value)

45

countByKey

#countByKey對(duì)Pair?RDD按key統(tǒng)計(jì)數(shù)量

pairRdd?=?sc.parallelize([(1,1),(1,4),(3,9),(2,16)])

pairRdd.countByKey()

defaultdict(int,?{1:?2,?3:?1,?2:?1})

saveAsTextFile

#saveAsTextFile保存rdd成text文件到本地

text_file?=?"./data/rdd.txt"

rdd?=?sc.parallelize(range(5))

rdd.saveAsTextFile(text_file)

#重新讀入會(huì)被解析文本

rdd_loaded?=?sc.textFile(file)

rdd_loaded.collect()

['2',?'3',?'4',?'1',?'0']

三，常用Transformation操作

Transformation轉(zhuǎn)換操作具有懶惰執(zhí)行的特性，它只指定新的RDD和其父RDD的依賴關(guān)系，只有當(dāng)Action操作觸發(fā)到該依賴的時(shí)候，它才被計(jì)算。

map

#map操作對(duì)每個(gè)元素進(jìn)行一個(gè)映射轉(zhuǎn)換

rdd?=?sc.parallelize(range(10),3)

rdd.collect()

[0,?1,?2,?3,?4,?5,?6,?7,?8,?9]

rdd.map(lambda?x:x**2).collect()

[0,?1,?4,?9,?16,?25,?36,?49,?64,?81]

filter

#filter應(yīng)用過(guò)濾條件過(guò)濾掉一些數(shù)據(jù)

rdd?=?sc.parallelize(range(10),3)

rdd.filter(lambda?x:x>5).collect()

[6,?7,?8,?9]

flatMap

#flatMap操作執(zhí)行將每個(gè)元素生成一個(gè)Array后壓平

rdd?=?sc.parallelize(["hello?world","hello?China"])

rdd.map(lambda?x:x.split("?")).collect()

[['hello',?'world'],?['hello',?'China']]

rdd.flatMap(lambda?x:x.split("?")).collect()

['hello',?'world',?'hello',?'China']

sample

#sample對(duì)原rdd在每個(gè)分區(qū)按照比例進(jìn)行抽樣,第一個(gè)參數(shù)設(shè)置是否可以重復(fù)抽樣

rdd?=?sc.parallelize(range(10),1)

rdd.sample(False,0.5,0).collect()

[1,?4,?9]

distinct

#distinct去重

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

rdd.distinct().collect()

[4,?1,?5,?2,?3]

subtract

#subtract找到屬于前一個(gè)rdd而不屬于后一個(gè)rdd的元素

a?=?sc.parallelize(range(10))

b?=?sc.parallelize(range(5,15))

a.subtract(b).collect()

[0,?1,?2,?3,?4]

union

#union合并數(shù)據(jù)

a?=?sc.parallelize(range(5))

b?=?sc.parallelize(range(3,8))

a.union(b).collect()

[0,?1,?2,?3,?4,?3,?4,?5,?6,?7]

intersection

#intersection求交集

a?=?sc.parallelize(range(1,6))

b?=?sc.parallelize(range(3,9))

a.intersection(b).collect()

[3,?4,?5]

cartesian

#cartesian笛卡爾積

boys?=?sc.parallelize(["LiLei","Tom"])

girls?=?sc.parallelize(["HanMeiMei","Lily"])

boys.cartesian(girls).collect()

[('LiLei',?'HanMeiMei'),

('LiLei',?'Lily'),

('Tom',?'HanMeiMei'),

('Tom',?'Lily')]

sortBy

#按照某種方式進(jìn)行排序

#指定按照第3個(gè)元素大小進(jìn)行排序

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

rdd.sortBy(lambda?x:x[2]).collect()

[(4,?1,?1),?(3,?2,?2),?(1,?2,?3)]

zip

#按照拉鏈方式連接兩個(gè)RDD，效果類似python的zip函數(shù)

#需要兩個(gè)RDD具有相同的分區(qū)，每個(gè)分區(qū)元素?cái)?shù)量相同

rdd_name?=?sc.parallelize(["LiLei","Hanmeimei","Lily"])

rdd_age?=?sc.parallelize([19,18,20])

rdd_zip?=?rdd_name.zip(rdd_age)

print(rdd_zip.collect())

[('LiLei',?19),?('Hanmeimei',?18),?('Lily',?20)]

zipWithIndex

#將RDD和一個(gè)從0開始的遞增序列按照拉鏈方式連接。

rdd_name?=??sc.parallelize(["LiLei","Hanmeimei","Lily","Lucy","Ann","Dachui","RuHua"])

rdd_index?=?rdd_name.zipWithIndex()

print(rdd_index.collect())

[('LiLei',?0),?('Hanmeimei',?1),?('Lily',?2),?('Lucy',?3),?('Ann',?4),?('Dachui',?5),?('RuHua',?6)]

四，常用PairRDD的轉(zhuǎn)換操作

PairRDD指的是數(shù)據(jù)為長(zhǎng)度為2的tuple類似(k,v)結(jié)構(gòu)的數(shù)據(jù)類型的RDD,其每個(gè)數(shù)據(jù)的第一個(gè)元素被當(dāng)做key，第二個(gè)元素被當(dāng)做value.

reduceByKey

#reduceByKey對(duì)相同的key對(duì)應(yīng)的values應(yīng)用二元?dú)w并操作

rdd?=?sc.parallelize([("hello",1),("world",2),

("hello",3),("world",5)])

rdd.reduceByKey(lambda?x,y:x+y).collect()

[('hello',?4),?('world',?7)]

groupByKey

#groupByKey將相同的key對(duì)應(yīng)的values收集成一個(gè)Iterator

rdd?=?sc.parallelize([("hello",1),("world",2),("hello",3),("world",5)])

rdd.groupByKey().collect()

[('hello',?),

('world',?)]

sortByKey

#sortByKey按照key排序,可以指定是否降序

rdd?=?sc.parallelize([("hello",1),("world",2),

("China",3),("Beijing",5)])

rdd.sortByKey(False).collect()

[('world',?2),?('hello',?1),?('China',?3),?('Beijing',?5)]

join

#join相當(dāng)于根據(jù)key進(jìn)行內(nèi)連接

age?=?sc.parallelize([("LiLei",18),

("HanMeiMei",16),("Jim",20)])

gender?=?sc.parallelize([("LiLei","male"),

("HanMeiMei","female"),("Lucy","female")])

age.join(gender).collect()

[('LiLei',?(18,?'male')),?('HanMeiMei',?(16,?'female'))]

leftOuterJoin和rightOuterJoin

#leftOuterJoin相當(dāng)于關(guān)系表的左連接

age?=?sc.parallelize([("LiLei",18),

("HanMeiMei",16)])

gender?=?sc.parallelize([("LiLei","male"),

("HanMeiMei","female"),("Lucy","female")])

age.leftOuterJoin(gender).collect()

[('LiLei',?(18,?'male')),?('HanMeiMei',?(16,?'female'))]

#rightOuterJoin相當(dāng)于關(guān)系表的右連接

age?=?sc.parallelize([("LiLei",18),

("HanMeiMei",16),("Jim",20)])

gender?=?sc.parallelize([("LiLei","male"),

("HanMeiMei","female")])

age.rightOuterJoin(gender).collect()

[('LiLei',?(18,?'male')),?('HanMeiMei',?(16,?'female'))]

cogroup

#cogroup相當(dāng)于對(duì)兩個(gè)輸入分別goupByKey然后再對(duì)結(jié)果進(jìn)行g(shù)roupByKey

x?=?sc.parallelize([("a",1),("b",2),("a",3)])

y?=?sc.parallelize([("a",2),("b",3),("b",5)])

result?=?x.cogroup(y).collect()

print(result)

print(list(result[0][1][0]))

[('a',?(,?)),?('b',?(,?))]

[1,?3]

subtractByKey

#subtractByKey去除x中那些key也在y中的元素

x?=?sc.parallelize([("a",1),("b",2),("c",3)])

y?=?sc.parallelize([("a",2),("b",(1,2))])

x.subtractByKey(y).collect()

[('c',?3)]

foldByKey

#foldByKey的操作和reduceByKey類似，但是要提供一個(gè)初始值

x?=?sc.parallelize([("a",1),("b",2),("a",3),("b",5)],1)

x.foldByKey(1,lambda?x,y:x*y).collect()

[('a',?3),?('b',?10)]

五，緩存操作

如果一個(gè)rdd被多個(gè)任務(wù)用作中間量，那么對(duì)其進(jìn)行cache緩存到內(nèi)存中對(duì)加快計(jì)算會(huì)非常有幫助。

聲明對(duì)一個(gè)rdd進(jìn)行cache后，該rdd不會(huì)被立即緩存，而是等到它第一次被計(jì)算出來(lái)時(shí)才進(jìn)行緩存。

可以使用persist明確指定存儲(chǔ)級(jí)別，常用的存儲(chǔ)級(jí)別是MEMORY_ONLY和EMORY_AND_DISK。

如果一個(gè)RDD后面不再用到，可以用unpersist釋放緩存，unpersist是立即執(zhí)行的。

緩存數(shù)據(jù)不會(huì)切斷血緣依賴關(guān)系，這是因?yàn)榫彺鏀?shù)據(jù)某些分區(qū)所在的節(jié)點(diǎn)有可能會(huì)有故障，例如內(nèi)存溢出或者節(jié)點(diǎn)損壞。

這時(shí)候可以根據(jù)血緣關(guān)系重新計(jì)算這個(gè)分區(qū)的數(shù)據(jù)。

#cache緩存到內(nèi)存中，使用存儲(chǔ)級(jí)別 MEMORY_ONLY。

#MEMORY_ONLY意味著如果內(nèi)存存儲(chǔ)不下，放棄存儲(chǔ)其余部分，需要時(shí)重新計(jì)算。

a?=?sc.parallelize(range(10000),5)

a.cache()

sum_a?=?a.reduce(lambda?x,y:x+y)

cnt_a?=?a.count()

mean_a?=?sum_a/cnt_a

print(mean_a)

#persist緩存到內(nèi)存或磁盤中，默認(rèn)使用存儲(chǔ)級(jí)別MEMORY_AND_DISK

#MEMORY_AND_DISK意味著如果內(nèi)存存儲(chǔ)不下，其余部分存儲(chǔ)到磁盤中。

#persist可以指定其它存儲(chǔ)級(jí)別，cache相當(dāng)于persist(MEMORY_ONLY)

from??pyspark.storagelevel?import?StorageLevel

a?=?sc.parallelize(range(10000),5)

a.persist(StorageLevel.MEMORY_AND_DISK)

sum_a?=?a.reduce(lambda?x,y:x+y)

cnt_a?=?a.count()

mean_a?=?sum_a/cnt_a

a.unpersist()?#立即釋放緩存

print(mean_a)

六，共享變量

當(dāng)spark集群在許多節(jié)點(diǎn)上運(yùn)行一個(gè)函數(shù)時(shí)，默認(rèn)情況下會(huì)把這個(gè)函數(shù)涉及到的對(duì)象在每個(gè)節(jié)點(diǎn)生成一個(gè)副本。

但是，有時(shí)候需要在不同節(jié)點(diǎn)或者節(jié)點(diǎn)和Driver之間共享變量。

Spark提供兩種類型的共享變量，廣播變量和累加器。

廣播變量是不可變變量，實(shí)現(xiàn)在不同節(jié)點(diǎn)不同任務(wù)之間共享數(shù)據(jù)。

廣播變量在每個(gè)機(jī)器上緩存一個(gè)只讀的變量，而不是為每個(gè)task生成一個(gè)副本，可以減少數(shù)據(jù)的傳輸。

累加器主要是不同節(jié)點(diǎn)和Driver之間共享變量，只能實(shí)現(xiàn)計(jì)數(shù)或者累加功能。

累加器的值只有在Driver上是可讀的，在節(jié)點(diǎn)上不可見。

#廣播變量?broadcast?不可變，在所有節(jié)點(diǎn)可讀

broads?=?sc.broadcast(100)

rdd?=?sc.parallelize(range(10))

print(rdd.map(lambda?x:x+broads.value).collect())

print(broads.value)

[100,?101,?102,?103,?104,?105,?106,?107,?108,?109]

100

#累加器?只能在Driver上可讀，在其它節(jié)點(diǎn)只能進(jìn)行累加

total?=?sc.accumulator(0)

rdd?=?sc.parallelize(range(10),3)

rdd.foreach(lambda?x:total.add(x))

total.value

45

#?計(jì)算數(shù)據(jù)的平均值

rdd?=?sc.parallelize([1.1,2.1,3.1,4.1])

total?=?sc.accumulator(0.1)

count?=?sc.accumulator(0)

def?func(x):

total.add(x)

count.add(1)

rdd.foreach(func)

total.value/count.value

2.625

七，分區(qū)操作

分區(qū)操作包括改變分區(qū)操作，以及針對(duì)分區(qū)執(zhí)行的一些轉(zhuǎn)換操作。

glom：將一個(gè)分區(qū)內(nèi)的數(shù)據(jù)轉(zhuǎn)換為一個(gè)列表作為一行。

coalesce：shuffle可選，默認(rèn)為False情況下窄依賴，不能增加分區(qū)。repartition和partitionBy調(diào)用它實(shí)現(xiàn)。

repartition：按隨機(jī)數(shù)進(jìn)行shuffle，相同key不一定在同一個(gè)分區(qū)

partitionBy：按key進(jìn)行shuffle，相同key放入同一個(gè)分區(qū)

HashPartitioner：默認(rèn)分區(qū)器，根據(jù)key的hash值進(jìn)行分區(qū)，相同的key進(jìn)入同一分區(qū)，效率較高，key不可為Array.

RangePartitioner：只在排序相關(guān)函數(shù)中使用，除相同的key進(jìn)入同一分區(qū)，相鄰的key也會(huì)進(jìn)入同一分區(qū)，key必須可排序。

TaskContext: ?獲取當(dāng)前分區(qū)id方法 TaskContext.get.partitionId

mapPartitions：每次處理分區(qū)內(nèi)的一批數(shù)據(jù)，適合需要分批處理數(shù)據(jù)的情況，比如將數(shù)據(jù)插入某個(gè)表，每批數(shù)據(jù)只需要開啟一次數(shù)據(jù)庫(kù)連接，大大減少了連接開支

mapPartitionsWithIndex：類似mapPartitions，提供了分區(qū)索引，輸入?yún)?shù)為(i，Iterator)

foreachPartition：類似foreach，但每次提供一個(gè)Partition的一批數(shù)據(jù)

glom

#glom將一個(gè)分區(qū)內(nèi)的數(shù)據(jù)轉(zhuǎn)換為一個(gè)列表作為一行。

a?=?sc.parallelize(range(10),2)

b?=?a.glom()

b.collect()

[[0,?1,?2,?3,?4],?[5,?6,?7,?8,?9]]

coalesce

#coalesce?默認(rèn)shuffle為False，不能增加分區(qū)，只能減少分區(qū)

#如果要增加分區(qū)，要設(shè)置shuffle?=?true

#parallelize等許多操作可以指定分區(qū)數(shù)

a?=?sc.parallelize(range(10),3)

print(a.getNumPartitions())

print(a.glom().collect())

3

[[0,?1,?2],?[3,?4,?5],?[6,?7,?8,?9]]

b?=?a.coalesce(2)

print(b.glom().collect())

[[0,?1,?2],?[3,?4,?5,?6,?7,?8,?9]]

repartition

#repartition按隨機(jī)數(shù)進(jìn)行shuffle，相同key不一定在一個(gè)分區(qū)，可以增加分區(qū)

#repartition實(shí)際上調(diào)用coalesce實(shí)現(xiàn)，設(shè)置了shuffle?=?True

a?=?sc.parallelize(range(10),3)

c?=?a.repartition(4)

print(c.glom().collect())

[[6,?7,?8,?9],?[3,?4,?5],?[],?[0,?1,?2]]

#repartition按隨機(jī)數(shù)進(jìn)行shuffle，相同key不一定在一個(gè)分區(qū)

a?=?sc.parallelize([("a",1),("a",1),("a",2),("c",3)])

c?=?a.repartition(2)

print(c.glom().collect())

[[('a',?1),?('a',?2),?('c',?3)],?[('a',?1)]]

partitionBy

#partitionBy按key進(jìn)行shuffle，相同key一定在一個(gè)分區(qū)

a?=?sc.parallelize([("a",1),("a",1),("a",2),("c",3)])

c?=?a.partitionBy(2)

print(c.glom().collect())

mapPartitions

#mapPartitions可以對(duì)每個(gè)分區(qū)分別執(zhí)行操作

#每次處理分區(qū)內(nèi)的一批數(shù)據(jù)，適合需要按批處理數(shù)據(jù)的情況

#例如將數(shù)據(jù)寫入數(shù)據(jù)庫(kù)時(shí)，可以極大的減少連接次數(shù)。

#mapPartitions的輸入分區(qū)內(nèi)數(shù)據(jù)組成的Iterator，其輸出也需要是一個(gè)Iterator

#以下例子查看每個(gè)分區(qū)內(nèi)的數(shù)據(jù),相當(dāng)于用mapPartitions實(shí)現(xiàn)了glom的功能。

a?=?sc.parallelize(range(10),2)

a.mapPartitions(lambda?it:iter([list(it)])).collect()

[[0,?1,?2,?3,?4],?[5,?6,?7,?8,?9]]

mapPartitionsWithIndex

#mapPartitionsWithIndex可以獲取兩個(gè)參數(shù)

#即分區(qū)id和每個(gè)分區(qū)內(nèi)的數(shù)據(jù)組成的Iterator

a?=?sc.parallelize(range(11),2)

def?func(pid,it):

s?=?sum(it)

return(iter([str(pid)?+?"|"?+?str(s)]))

[str(pid)?+?"|"?+?str]

b?=?a.mapPartitionsWithIndex(func)

b.collect()

#利用TaskContext可以獲取當(dāng)前每個(gè)元素的分區(qū)

from?pyspark.taskcontext?import?TaskContext

a?=?sc.parallelize(range(5),3)

c?=?a.map(lambda?x:(TaskContext.get().partitionId(),x))

c.collect()

[(0,?0),?(1,?1),?(1,?2),?(2,?3),?(2,?4)]

foreachPartitions

#foreachPartition對(duì)每個(gè)分區(qū)分別執(zhí)行操作

#范例：求每個(gè)分區(qū)內(nèi)最大值的和

total?=?sc.accumulator(0.0)

a?=?sc.parallelize(range(1,101),3)

def?func(it):

total.add(max(it))

a.foreachPartition(func)

total.value

199.0

aggregate

#aggregate是一個(gè)Action操作

#aggregate比較復(fù)雜，先對(duì)每個(gè)分區(qū)執(zhí)行一個(gè)函數(shù)，再對(duì)每個(gè)分區(qū)結(jié)果執(zhí)行一個(gè)合并函數(shù)。

#例子：求元素之和以及元素個(gè)數(shù)

#三個(gè)參數(shù)，第一個(gè)參數(shù)為初始值，第二個(gè)為分區(qū)執(zhí)行函數(shù)，第三個(gè)為結(jié)果合并執(zhí)行函數(shù)。

rdd?=?sc.parallelize(range(1,21),3)

def?inner_func(t,x):

return((t[0]+x,t[1]+1))

def?outer_func(p,q):

return((p[0]+q[0],p[1]+q[1]))

rdd.aggregate((0,0),inner_func,outer_func)

(210,?20)

aggregateByKey

#aggregateByKey的操作和aggregate類似，但是會(huì)對(duì)每個(gè)key分別進(jìn)行操作

#第一個(gè)參數(shù)為初始值，第二個(gè)參數(shù)為分區(qū)內(nèi)歸并函數(shù)，第三個(gè)參數(shù)為分區(qū)間歸并函數(shù)

a?=?sc.parallelize([("a",1),("b",1),("c",2),

("a",2),("b",3)],3)

b?=?a.aggregateByKey(0,lambda?x,y:max(x,y),

lambda?x,y:max(x,y))

b.collect()

[('b',?3),?('a',?2),?('c',?2)]

總結(jié)

以上是生活随笔為你收集整理的spark算子大全glom_2小时入门Spark之RDD编程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。