PySpark のペア RDD を使う

PySparkで学ぶBig Data入門

Upendra Devisetty

Science Analyst, CyVerse

PySpark におけるペア RDD の導入

実データは通常キー/値の組です
各行はキーで、1つ以上の値に対応します
ペア RDD はこの種のデータに適した構造です
ペア RDD: キーは識別子、値はデータ

ペア RDD の作成

ペア RDD の一般的な作成方法は2つ
- キー/値タプルのリストから
- 通常の RDD から
データをキー/値形式にしてペア RDD を作る

my_tuple = [('Sam', 23), ('Mary', 34), ('Peter', 25)]
pairRDD_tuple = sc.parallelize(my_tuple)

my_list = ['Sam 23', 'Mary 34', 'Peter 25']
regularRDD = sc.parallelize(my_list)
pairRDD_RDD = regularRDD.map(lambda s: (s.split(' ')[0], s.split(' ')[1]))

ペア RDD の変換

通常の変換はすべてペア RDD でも動作します
個々の要素ではなくキー/値に対する関数を渡します
ペア RDD の変換例
- reduceByKey(func): 同じキーの値を集約
- groupByKey(): 同じキーの値をグループ化
- sortByKey(): キーで並べ替えた RDD を返す
- join(): キーに基づき2つのペア RDD を結合

reduceByKey() 変換

reduceByKey() は同じキーの値を結合します
データセット内の各キーに対し並列で実行されます
これはアクションではなく変換です

regularRDD = sc.parallelize([("Messi", 23), ("Ronaldo", 34), 
                             ("Neymar", 22), ("Messi", 24)])
pairRDD_reducebykey = regularRDD.reduceByKey(lambda x,y : x + y)
pairRDD_reducebykey.collect()

[('Neymar', 22), ('Ronaldo', 34), ('Messi', 47)]

sortByKey() 変換

sortByKey() はキーでペア RDD を並べ替えます
昇順または降順でキー順の RDD を返します

pairRDD_reducebykey_rev = pairRDD_reducebykey.map(lambda x: (x[1], x[0]))
pairRDD_reducebykey_rev.sortByKey(ascending=False).collect()

[(47, 'Messi'), (34, 'Ronaldo'), (22, 'Neymar')]

groupByKey() 変換

groupByKey() はペア RDD {{1}} で同じキーの値をまとめます

airports = [("US", "JFK"),("UK", "LHR"),("FR", "CDG"),("US", "SFO")]
regularRDD = sc.parallelize(airports)
pairRDD_group = regularRDD.groupByKey().collect()
for cont, air in pairRDD_group:
  print(cont, list(air))

FR ['CDG']
US ['JFK', 'SFO']
UK ['LHR']

join() 変換

join() 変換はキーに基づいて2つのペア RDD を結合します

RDD1 = sc.parallelize([("Messi", 34),("Ronaldo", 32),("Neymar", 24)])
RDD2 = sc.parallelize([("Ronaldo", 80),("Neymar", 120),("Messi", 100)])

RDD1.join(RDD2).collect()

[('Neymar', (24, 120)), ('Ronaldo', (32, 80)), ('Messi', (34, 100))]

Ayo berlatih!

PySparkで学ぶBig Data入門