Spark Dataframe Operasyonları-1: İndeks Ekleme ve Sütun Sırasını ve Türünü Değiştirme

Erkan ŞİRİN 5 Kasım 2017 Büyük Veri, Scala, Spark 0 3346 961 kelime - 6 dakika94

---

Bu yazıya puan ver

(No Ratings Yet)

Loading...

---

Merhaba, bu yazımızda veri ön hazırlığı aşamasında veri yüklerken, Spark Dataframe oluştururken, dataframe şekillendirirken, onu dönüştürürken, dataframe üzerinde veri keşfi yaparken vb. işlemler için kullandığım örnek uygulamaları paylaşacağım. Basit bir yazı olacak ancak bunu önemsiyorum, çünkü birçok insan veri yükleme ve veri hazırlığı esnasında sorun yaşıyor. Spark 2.0 sürümüyle birlikte Spark dataframe API’ye daha fazla önem vermeye başladı, ilişkisel dünyanın gücünü farketti sanırım 🙂 Sizlere tavsiyem Spark’a yeni giriyorsanız öncelikle Spark RDD ile değil Dataframe ile başlayın.

Bu yazıda kullanacağım araçlar şunlardır:

• Spark Sürümü: Spark 2.1.1
• Geliştirme Ortamı: Apache Zeppelin Notebook
• Kaynak Yönetimi: Apache YARN (Yani Spark’ı YARN modunda çalıştırıyorum)
• Veri Depolama: Hadoop HDFS
• Programlama Dili: Scala

Dataframe Indeks Ekleme

Notasyon basit; verimizi okurken sonuna withColumn(“id için sütun adı”, monotonically_increasing_id()) eklediğimizde bize 0’dan başlayıp satır sayısına kadar artan long tipinde yeni bir sütun ekler. Şayet indeksi 0’dan değilde 1’den başlatmak istiyorsak monotonically_increasing_id() yanına +1 yazabiliriz. Şimdi myrange adında 1’den 100’e kadar ikişer artan tek sütunlu dataframe için 1’den başlayan bir indeks ekleyeceğim. monotonicallyIncreasingId kullanmak için sql.functions kütüphanesini indirmeliyiz. Ancak bir hatırlatma yapmak istiyorum. Bu fonksiyon indeksin artışını ve benzersizliğini garanti etse de ardışık olmayı garanti etmez.

import org.apache.spark.sql.functions._
val myrange = spark.range(1,101,2).toDF("number").withColumn("id", monotonically_increasing_id()+1)
myrange.printSchema()
myrange.show()

import org.apache.spark.sql.functions._
myrange: org.apache.spark.sql.DataFrame = [number: bigint, id: bigint]
root
 |-- number: long (nullable = false)
 |-- id: long (nullable = false)
+------+---+
|number| id|
+------+---+
|     1|  1|
|     3|  2|
|     5|  3|
|     7|  4|
|     9|  5|
|    11|  6|
|    13|  7|
|    15|  8|
|    17|  9|
|    19| 10|
+------+---+
only showing top 10 rows


Sonuç:

myrange.distinct().count()
myrange.select('id).distinct().count()

Her satırın benzersiz bir rakam alıp almadığının sağlamasını dataframe satırlarını ve yeni eklediğimiz indeks sütununu saydırarak yapabiliriz.

res75: Long = 50
res76: Long = 50

Sonuç:

Sonucun aynı olması, fonksiyonun doğru çalıştığını gösterir.

Spark Dataframe Sütun Sırasını/Dizilimini Değiştirmek

Spark Dataframe’de sütun isimlerinin dizilimini değiştirmek isteyebiliriz. Bunu yapmak pandas kütüphanesi kadar kolay değil, çünkü herşey tek bir makinenin ana belleğinde olup bitmiyor, ancak bunun da bir yöntemi var üstelik çok da karmaşık değil. Örneğin yukarıdaki basit dataframede id ile number sütunlarını yer değiştirelim.
Önce mevcut dataframe sütun isimlerini yazdıralım. Çünkü buradan kopyalayıp sırasını değiştireceğiz. Burada iki sütun olduğu için bu iş gereksiz görünebilir ancak sütun sayısı fazla olduğunda çok işe yarayacaktır.

myrange.columns
res85: Array[String] = Array(number, id)

Yukarıdaki Array içini kopyalayıp yeni bir dizi oluşturalım ancak sütun isimlerini yer değiştirelim ve liste elemanlarını çift tırnak içine alalım.

var new_columns = Seq("id", "number")

Yeni dataframe için myrange2 kullanalım ve yeni sütun isimleri ile dataframe oluşturalım.

val myrange2 = myrange.toDF(new_columns:_*)

myrange2.show(5)
 +---+------+
| id|number|
+---+------+
|  1|     1|
|  2|     3|
|  3|     5|
|  4|     7|
|  5|     9|
+---+------+
only showing top 5 rows

Sonucu doğrulamak için yeni dataframe’in beş satırını gösterelim.

Evet görüldüğü gibi id sütunu başa geldi. Burada şunu belirtmek isterim Spark Dataframe immutable olduğu için dataframe yapısı değiştikçe yeni bir isimle tekrar atama yapmalıyız, myrange2 yaratmamızın sebebi budur.

Spark Dataframe when() Fonksiyonu ile Yeni Sütun Ekleme

Veri ön hazırlığı kapsamında bazen cinsiyet gibi iki veye daha çok sınıflı nitelikleri nümerik değerler ile değiştirmek isteriz. Şimdi yukarıdaki myrange2 üzerinde when() fonksiyonu ile koşullu değişim yapacağız. Yeni bir sütuna id sütunundaki değerleri tek ise tek çift ise çift olarak yazacağız. Bunun için de when() fonksiyonu ve modu kullanacağız. Mod kullanmak için eşitlikte % kullanmamız yeterlidir. Aşağıda yapılan bir bakıma if, else ile yapılan koşul sınamasıdır. Eğer böyleyse şunu yap, yok değilse bunu yap.

val myrange3 = myrange2.select('id, 'number, when(myrange2("id") %2 === 0, "çift")
  .when(myrange2("id") %2 === 1, "tek")
  .otherwise("Bilinmiyor").as("TekCift"))

Oluşuan yeni myrange3’ü show fonksiyonu ile görelim:

myrange3.show(5)
 +---+------+-------+
| id|number|TekCift|
+---+------+-------+
|  1|     1|    tek|
|  2|     3|   çift|
|  3|     5|    tek|
|  4|     7|   çift|
|  5|     9|    tek|
+---+------+-------+
only showing top 5 rows

Evet tam istediğimiz gibi, id sütununda tek rakamların karşısına tek, çift rakamların karşısına da çift yazdırdık.

Spark Dataframe Sütun Türünü Değiştirme

Şimdi de sütun türünü değiştirelim. Öncelikle mevcut dataframe şemasını bir yazdıralım:

myrange3.printSchema()
root
 |-- id: long (nullable = false)
 |-- number: long (nullable = false)
 |-- TekCift: string (nullable = false)

number sütunu long, onu string yapalım. Bunun için SQL’den bildiğimiz cast() fonksiyonunu kullanacağız.

val myrange4 = myrange3.withColumn("number",$"number".cast("string"))
myrange4.printSchema()
root
 |-- id: long (nullable = false)
 |-- number: string (nullable = false)
 |-- TekCift: string (nullable = false)

Evet yukarıda gördüğümüz gibi number sütunu long türünden string türüne döndü. Dikkat ettiyseniz withCloumn() kullandığımız halde sütun sayımız artmadı üçte kaldı. Bunun nedeni yeni sütun adını eski sütun adıyla aynı vermemizdir.