#spark action

Aksiyonları RDD’ ler üzerinde hesaplamalar yapmak için kullanılır. En sıklıkla kullanılan aksiyon reduce() iki eleman üzerinde çalışarak aynı türde yeni bir değer üretir. Örneğin + gibi bir fonksiyon ile tüm değerleri toplayabilir, toplam kayıt sayısını alabilir farklı türde hesaplama fonksiyonları kullanabiliriz.

Örneğin Python’ da: toplam=RDD.reduce(lambda x,y: x+y) Benzer şekilde Java’da: Integer toplam=RDD.reduce(new Function2<Integer,Integer>() { public Integer call(Integer x, Integer y) { return x+y; });

Fold() aksiyonu reduce()’ e benzer biçimde kullanılır. Fakat reduce’ den farklı olarak bir başlangıç “sıfır” değeri ile çalışır. Bu “sıfır” değer toplama için sıfır (0) çarpma için ise bir (1) gibi etkisiz eleman olmalıdır. Bu başlangıç değeri fold()’ un boş liste üzerinde’ de hatasız çalışmasını sağlar. Eğer hiç elemanı olmayan bir RDD üzerinde reduce() çalıştırırsak “null” değerler üzerinde işlemler programın kırılmasına neden olur.

Reduce() ve fold() sonuç olarak RDD ile aynı türde/tipte değer değer geriye döndürür. Bu sum gibi bir operasyonda sorun teşkil etmez ancak bazı durumlarda hesaplanan sonucun farklı türde olması gerekebilir. Örneğin tüm tamsayı değerin ortalamasının alınması operasyonunda olduğu gibi.  Ortalama hesaplamaya çalıştığımız durumlarda, hem toplamı hesaplamalı hem de kaç verimiz (count) olduğunu saklayabilmeliyiz. Bu durumda iki farklı yöntem uygulamak mümkündür. İlk yöntem map() ile her elemanı ayırıp yanına sayı olarak “1” oluşturmak ve (4,1), (8,1), (-3,1) gibi çiftler üretmektir. Ardından reduce() ile bu çiftler üzerinde çalışıp hem count hem de toplamı bulmak mümkündür.

4+8+(-3) = 9 1+1+1=3 9/3= 3 ortalama bulunur.

Diğer yöntem ise aggregate() fonksiyonunu kullanarak kaynak RDD’ den farklı türde hesaplama yapmaktır. Aggregate() aksiyonunda da tıpkı fold’ da olduğu gibi bir başlangıç “sıfır” değeri (0,1) verilir. Fakat bu başlangıç sıfır değeri kaynak RDD (int) ‘ den farklı (double) olabilir. Ayrıca RDD’ deki elemanları birleştirmek için bir fonksiyon adı parametre olarak verilir. Son adımda ise ikinci bir fonksiyon ile farklı düğümlerde yapılan hesaplamaların birleştirilmesi sağlanır.

Sık Kullanılan Aksiyonlar:

RDD={1,2,3,3} .collect() tüm RDD nin ana düğüme getirilmesini sağlar .count() RDD’nin eleman sayısını getirir (4). .countByValue() RDD’ nin her elemanın sayısını ayrı ayrı getirir. { (1,1),(2,1),(3,2)} .take(sayı) örneğin take(2) RDD’ den verilen sayı kadar değer döndürür. take(2)-> {1,2} .takeOrdered(sayı(order) RDD.takeOrdered(2)(order)-> {3,3} .reduce(function) ile tüm elemanlar paralel işlenir RDD.reduce(/x,y)->x+y)=9 RDD.fold((0)(x,y)->x+y)=9 RDD.aggregate(zeroVal, operation, combination)- RDD.aggregate((0,0),(x,y)->(x._1+y,x._2+1),(x,y)->(x._1+y._1, x._2+y._2))

Bir sonraki yazımda kısaca Spark Cache  (Persistance) kavramlarına kısa bir giriş yaptıktan sonra, belki de en önemli konumuz olan KEY/VALUE çiftleri ile çalışmayı anlatmayı planlıyorum.

RDD’ler üzerinde gerçekleştirilen bir operasyon sonunda yeni bir RDD elde ettiğimiz operasyonlara transfer/dönüşüm (transformation) operasyonları adı verilir. Tamamı için geçerli olmasa da, çoğu dönüşüm RDD üzerindeki her bir veri üzerinde tek tek çalışır. Dönüşümler tembel (lazy) olarak hesaplanır. Yani dönüşüm işlemi bir aksiyon (action) uygulandığında gerçekleştirilir. Bir RDD üzerinde gerçekleştirilen dönüşüm, kaynak RDD ‘ yi değiştirmez. Farklı yeni bir RDD nesnesi oluşturulur.  (Ancak hesaplanmaz!) Kaynak RDD başka operasyonlar için kullanılabilir. Örneğin: inputRDD=sc.textFile(“log.tct”) errorsRDD=inputRDD.filter(lambda x:”error” in x) warningsRDD=inputRDD.filter(lambda x:”error” in x) errAndwarnRDD=errorsRDD.union(warningsRDD)

Yukarıdaki örnekte inputRDD kullanılarak iki farklı RDD oluşturulmuştur. Union ile bu iki farklı RDD birleştirilmiştir. Aslında Union yerine aynı anda he error hem de warningleri filtreleyen tek bir lambda yazmak mümkündür. Fakat Union “lineage” kavramının anlaşılması açısından önemlidir. Spark RDD’ ler arasındaki tüm ilişki ve bağımlılıkları saklar. Bu ilişkileri gösteren graf’a lineage (bağımlılık) graf adı verilir.

Dönüşüm sonrasında oluşan RDD’ler üzerinde çalıştırılabilecek iki önemli operasyon take() ve collect() aksiyonlarıdır. Take ile büyük bir veri setinden oluşan RDD içerisinden küçük/az sayıda örnek almak mümkündür. Python: inputRDD=sc.textFile(“log.tct”) errorsRDD=inputRDD.filter(lambda x:”error” in x) warningsRDD=inputRDD.filter(lambda x:”error” in x) errAndwarnRDD=errorsRDD.union(warningsRDD) print “Filtrelenmiş RDD “ + errAndwarnRDD.count() + “ satirdan oluşmaktadır.” print “Hatalı satırların 10 tanesi:” for satir in errAndwarnRDD.take(10) print satir

Aynı örnek Java ile geliştirildiğinde (önceki yazılarımızdan  RDD nin javada nasıl oluşturulduğuna bakabilirsiniz) System.out.Println(“ Hatali satirlarin 10 tanesi…”); for(String satir: errAndwarnRDD.take(10)) System.out.Println(satir);