PySpark'ta Çift RDD'lerle Çalışma

PySpark ile Big Data Temelleri

Upendra Devisetty

Science Analyst, CyVerse

PySpark'ta çift RDD'lere giriş

Gerçek veri kümeleri genelde anahtar/değer çiftleridir
Her satır bir anahtardır ve bir veya daha çok değere eşlenir
Çift RDD, bu tür verilerle çalışmak için özel bir yapıdır
Çift RDD: Anahtar tanımlayıcıdır, değer veridir

Çift RDD oluşturma

Çift RDD oluşturmanın iki yaygın yolu
- Anahtar-değer demetinden
- Normal bir RDD'den
Veriyi çift RDD için anahtar/değer biçimine getirin

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]))

Çift RDD'lerde dönüşümler

Tüm normal dönüşümler çift RDD üzerinde çalışır
Tek tek öğeler yerine anahtar-değer çiftleri üzerinde çalışan fonksiyonlar geçirmelisiniz
Çift RDD dönüşümlerine örnekler
- reduceByKey(func): Aynı anahtara sahip değerleri birleştirir
- groupByKey(): Aynı anahtardaki değerleri gruplar
- sortByKey(): Anahtara göre sıralı bir RDD döndürür
- join(): İki çift RDD'yi anahtara göre birleştirir

reduceByKey() dönüşümü

reduceByKey() dönüşümü, aynı anahtardaki değerleri birleştirir
Veri kümesindeki her anahtar için paralel çalışır
Bir dönüşümdür, eylem değildir

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() dönüşümü

sortByKey() işlemi, çift RDD'yi anahtara göre sıralar
Artan ya da azalan anahtara göre sıralı bir RDD döndürür

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() dönüşümü

groupByKey(), çift RDD {{1}} içinde aynı anahtara sahip tüm değerleri gruplar

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() dönüşümü

join() dönüşümü, iki çift RDD'yi anahtarlarına göre birleştirir

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))]

Hadi pratik yapalım

PySpark ile Big Data Temelleri