Apache Spark Eğitimi: PySpark ile ML

pip ile pyspark İndirme

Ardından, pip yardımıyla PySpark’ı indirip kurabilirsiniz. Bu oldukça kolaydır ve başka herhangi bir paketi kurmaya benzer. Alışık olduğunuz komutu çalıştırırsınız ve ağır işi sizin yerinize halleder:

$ pip install pyspark

Alternatif olarak Spark indirme sayfasına da gidebilirsiniz. İlk üç adımda varsayılan seçenekleri koruyun; 4. adımda indirilebilir bir bağlantı bulacaksınız. İndirmek için o bağlantıya tıklayın. Bu eğitim için 2.2.0 Spark Sürümü ve “Apache Hadoop 2.7 ve sonrası için önceden derlenmiş” paket türünü indireceksiniz.

Not indirme işleminin tamamlanması biraz zaman alabilir!

Homebrew ile Spark İndirme

Spark’ı Homebrew, ücretsiz ve açık kaynaklı bir paket yöneticisi, ile de kurabilirsiniz. Bu, özellikle macOS ile çalışıyorsanız kullanışlıdır.

Kişisel bilgisayarınıza Spark’ı aramak, daha fazla bilgi almak ve son olarak kurmak için şu komutları çalıştırmanız yeterlidir:

# Search for spark
$ brew search spark

# Get more information on apache-spark
$ brew info apache-spark

# Install apache-spark
$ brew install apache-spark

Spark’ı İndirme ve Kurma

Sonraki adımda, Downloads klasörünüzde beliren dizinin arşivini açtığınızdan emin olun. Bunu, spark-2.2.0-bin-hadoop2.7.tgz arşivine çift tıklayarak otomatik olarak veya Terminal’i açıp aşağıdaki komutu çalıştırarak yapabilirsiniz:

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgz

Ardından, arşivden çıkarılan klasörü aşağıdaki satırı çalıştırarak /usr/local/spark konumuna taşıyın:

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark

Not bu klasörü yeni konuma taşımak için izninizin reddedildiğini belirten bir hata alırsanız, bu komutun önüne sudo eklemelisiniz. Yukarıdaki satır bu durumda $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark olur. Sizden parolanız istenecektir; bu genellikle bilgisayarınızı açarken kilidini kaldırmak için kullandığınız paroladır :)

Artık her şey hazır olduğuna göre, /usr/local/spark yolundaki README dosyasını açın. Bunu şu komutu çalıştırarak yapabilirsiniz

$ cd /usr/local/spark

Bu sizi olmanız gereken klasöre getirir. Ardından klasörü incelemeye ve içinde yer alan README dosyasını okumaya başlayabilirsiniz.

Önce bu spark klasöründeki dosya ve klasörlerin listesini almak için $ ls kullanın. İçeride bir README.md dosyası olduğunu göreceksiniz. Aşağıdaki komutlardan biriyle açabilirsiniz:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

İpucu dosya adını yazarken otomatik tamamlamak için klavyenizdeki sekme tuşunu kullanın :) Bu size zaman kazandıracaktır.

Bu README’nin size Spark hakkında genel bilgiler, çevrimiçi dokümantasyon, Spark’ı derleme, Etkileşimli Scala ve Python kabukları, örnek programlar ve daha fazlasını sunduğunu göreceksiniz.

Burada en çok ilginizi çekebilecek şey Spark’ın nasıl derleneceği bölümü olabilir; ancak bunun yalnızca önceden derlenmiş bir sürüm indirmediyseniz özellikle ilgili olduğunu unutmayın. Bu eğitimde ise önceden derlenmiş bir sürüm indirdiniz. README’den çıkmak için CTRL + X’e basabilirsiniz; bu sizi spark klasörüne geri getirir.

Henüz derlenmemiş bir sürüm seçtiyseniz, README dosyasında listelenen komutu çalıştırdığınızdan emin olun. Bu metin yazılırken komut şu şekildedir:

$ build/mvn -DskipTests clean package run

Bu komutun çalışmasının biraz zaman alabileceğini unutmayın.

Spark Uygulamaları ve Spark Kabuk

Etkileşimli kabuk, bir Read-Eval(uate)-Print-Loop (REPL) ortamı örneğidir; yani yazdığınız her şey okunur, değerlendirilir ve çıktısı verilir; böylece analizine devam edebilirsiniz. Bu size Jupyter ile çalışırken tanıyabileceğiniz, güçlü bir etkileşimli Python kabuğu olan IPython’ı hatırlatabilir. Daha fazla bilgi isterseniz DataCamp’in IPython mı Jupyter mi blog yazısını okuyun.

Bu, ihtiyacınız olan tüm etkileşimli çalışmalar için Python’un yanı sıra Scala’da da mevcut olan kabuğu kullanabileceğiniz anlamına gelir.

Bu kabuğa ek olarak Spark uygulamaları da yazabilir ve dağıtabilirsiniz. Spark uygulamaları yazmanın aksine, kabukta SparkSession sizin için zaten oluşturulmuştur; böylece doğrudan çalışmaya başlayabilir ve bir tane oluşturmaya zaman harcamazsınız.

Şimdi merak ediyor olabilirsiniz: SparkSession nedir?

Bu, Spark işlevselliğinin ana giriş noktasıdır: bir Spark kümesine bağlantıyı temsil eder ve bu kümede RDD’ler oluşturmak ve değişkenleri yayınlamak için kullanılabilir. Spark ile çalışırken her şey bu SparkSession ile başlar ve biter. Not Spark 2.0.0’dan önce, üç ana bağlantı nesnesi SparkContext, SqlContext ve HiveContext idi.

Daha fazlasını ileride göreceksiniz. Şimdilik kabuğa odaklanalım.

Python Spark Kabuk

/usr/local/spark konumundaki spark klasörünün içinden

$ ./bin/pyspark

komutunu çalıştırabilirsiniz. İlk başta bazı metinler görünür. Ardından aşağıdakine benzer biçimde “Spark” ifadesini görürsünüz:

Python 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016, 12:39:47) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
17/07/26 11:41:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/26 11:41:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/

Using Python version 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016 12:39:47)
SparkSession available as 'spark'.
>>>

Bunu gördüğünüzde, etkileşimli kabukta denemeler yapmaya hazırsınız demektir!

İpucu: Spark kabuğu yerine IPython kabuğunu kullanmayı tercih ediyorsanız, aşağıdaki ortam değişkenini ayarlayarak bunu yapabilirsiniz:

export PYSPARK_DRIVER_PYTHON="/usr/local/ipython/bin/ipython"

>>> rdd1 = spark.sparkContext.parallelize([('a',7),('a',2),('b',2)])
>>> rdd2 = spark.sparkContext.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])
>>> rdd3 = spark.sparkContext.parallelize(range(100))

>>> rdd1.reduce(lambda a,b: a+b)

>>> rdd2.flatMapValues(lambda x: x).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

Jupyter Notebook’ta PySpark

Bu eğitimin bu bölümünde ishell’i kullanmayacak, kendi uygulamanızı oluşturacaksınız. Bunu bir Jupyter Notebook’ta yapacaksınız. İhtiyacınız olan her şeyi zaten kurdunuz; dolayısıyla PySpark’ı Jupyter’da çalıştırmak için fazla bir şey yapmanız gerekmeyecek.

Notebook uygulamasını her zamanki gibi $ jupyter notebook komutunu çalıştırarak başlatabilirsiniz. Sonra yeni bir notebook oluşturup findspark kütüphanesini içe aktarırsınız ve init() işlevini kullanırsınız. Bu durumda, Spark’ı bu yola kurduğunuzdan emin olduğunuz için init() işlevine /usr/local/spark yolunu vereceksiniz.

# Import findspark 
import findspark

# Initialize and provide path
findspark.init("/usr/local/spark")

# Or use this alternative
#findspark.init()

İpucu: Yolunuzun doğru ayarlanıp ayarlanmadığına dair bir fikriniz yoksa veya Spark’ı bilgisayarınıza nereye kurduğunuzu bilmiyorsanız, Spark’ın kurulu olduğu konumu otomatik olarak algılamak için her zaman findspark.find() kullanabilirsiniz.

Jupyter’da Spark ile çalışmanın alternatif yollarını arıyorsanız, Apache Spark in Python: Beginner’s Guide içeriğimize bakın.

Tüm bunları hallettiğinize göre, artık ilk Spark programınızı oluşturmaya başlayabilirsiniz!

İlk Spark Programınızı Oluşturma

İlk olarak pyspark paketinden SparkContext’i içe aktaracak ve başlatacaksınız. Unutmayın bunu daha önce yapmanıza gerek yoktu; çünkü etkileşimli Spark kabuğu sizin için bunu otomatik olarak oluşturup başlatıyordu! Burada biraz daha fazla işi kendiniz yapmanız gerekecek :)

pyspark.sql içinden SparkSession modülünü içe aktarın ve builder() yöntemiyle bir SparkSession oluşturun. Sonrasında bağlanılacak ana URL’yi, uygulama adını ayarlayabilir, yürütücü belleği gibi ek yapılandırmalar ekleyebilir ve son olarak getOrCreate() ile mevcut Spark oturumunu alabilir veya çalışmıyorsa bir tane oluşturabilirsiniz.

# Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContext

Not “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’” benzeri bir FileNotFoundError alırsanız, Spark PATH’inizi (yeniden) ayarlamanız gerektiğini bilirsiniz. $ cd komutunu çalıştırarak ana dizininize gidin ve ardından $ nano .bash_profile komutuyla .bash_profile dosyasını düzenleyin.

Dosyanın sonuna aşağıdakine benzer bir şey ekleyin

export SPARK_HOME="/usr/local/spark"

Dosyadan çıkmak için CTRL + X kullanın; ancak değişiklikleri kaydettiğinizden emin olmak için değişiklikleri onaylamak üzere Y girin. Ardından, source .bash_profile komutunu çalıştırarak değişiklikleri etkinleştirmeyi unutmayın.

İpucu: isterseniz ek ortam değişkenleri de ayarlayabilirsiniz; Muhtemelen gerek duymazsınız ama ihtiyaç halinde ayarlayabileceğinizi bilmek iyi olur. Şu örneklere bakın:

# Set a fixed value for the hash seed secret
export PYTHONHASHSEED=0

# Set an alternate Python executable
export PYSPARK_PYTHON=/usr/local/ipython/bin/ipython

# Augment the default search path for shared libraries
export LD_LIBRARY_PATH=/usr/local/ipython/bin/ipython

# Augment the default search path for private libraries 
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-*-src.zip:$PYTHONPATH:$SPARK_HOME/python/

Not ayrıca şu anda varsayılan bir SparkSession başlattınız. Ancak çoğu durumda bunu daha fazla yapılandırmak istersiniz. Büyük veriyle çalışmaya başladığınızda bunun gerçekten gerekli olduğunu göreceksiniz. Daha fazla bilgi için bu sayfaya bakın.

Verinizi Yükleme

Bu eğitim California Housing veri setini kullanır. 1997 tarihli, Pace, R. Kelley ve Ronald Barry tarafından yazılan ve Statistics and Probability Letters dergisinde yayımlanan Sparse Spatial Autoregressions başlıklı bir makalede yer almıştır. Araştırmacılar bu veri setini 1990 California nüfus sayımı verilerini kullanarak oluşturmuşlardır.

Veri, her nüfus sayımı blok grubu için bir satır içerir. Blok grubu, ABD Nüfus Sayım Bürosu’nun örnek verileri yayımladığı en küçük coğrafi birimdir (bir blok grubu tipik olarak 600 ila 3.000 kişi arasında nüfusa sahiptir). Bu örnekte bir blok grubu ortalama olarak coğrafi olarak kompakt bir bölgede yaşayan 1425,5 birey içerir. Bu bilgileri bu web sayfasından veya yukarıda belirtilen ve burada bulabileceğiniz makaleyi okuyarak edinebilirsiniz.

Bu mekânsal veriler, konut fiyatlarına ilişkin 20.640 gözlem ve 9 ekonomik değişken içerir:

• Boylam (Longitude), her blok grubu için yeryüzü ekvatorunun kuzey veya güney yönündeki açısal uzaklığa karşılık gelir;
• Enlem (Latitude), her blok grubu için yeryüzü ekvatorunun doğu veya batı yönündeki açısal uzaklığa karşılık gelir;
• Ortalama konut yaşı (Housing median age), bir blok grubuna ait kişilerin medyan yaşıdır. Not medyan, gözlenen değerlerin frekans dağılımının orta noktasında yer alan değerdir;
• Toplam oda (Total rooms), her blok grubundaki evlerdeki toplam oda sayısıdır;
• Toplam yatak odası (Total bedrooms), her blok grubundaki evlerdeki toplam yatak odası sayısıdır;
• Nüfus (Population), bir blok grubunun sakin sayısıdır;
• Hane sayısı (Households), blok grup başına konut birimleri ve sakinlerini ifade eder;
• Ortalama gelir (Median income), bir blok grubuna ait kişilerin medyan gelirini kaydetmek için kullanılır; ve,
• Ortalama konut değeri (Median house value), bağımlı değişkendir ve blok grup başına medyan konut değerini ifade eder.

Ayrıca, bağımsız ve bağımlı değişkenler için sıfır girişleri olan tüm blok gruplarının veriden çıkarıldığını da öğrenirsiniz.

Ortalama konut değeri bağımlı değişkendir ve ML modelinizde hedef değişken rolünü alacaktır.

Veriyi buradan indirebilirsiniz. houses.zip klasörünü bulun, indirin ve arşivden çıkararak veri klasörlerine erişin.

Sonraki adımda, indirdiğiniz klasördeki verileri RDD’lere okumak için textFile() yöntemini kullanacaksınız. Bu yöntem, dosya için bir URI alır; bu durumda makinenizin yerel yoludur ve onu satırlar koleksiyonu olarak okur. Kolaylık olması açısından yalnızca .data dosyasını değil, başlığı içeren .domain dosyasını da okuyacaksınız. Bu, değişkenlerin sırasını tekrar kontrol etmenizi sağlar.

# Load in the data
rdd = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.data')

# Load in the header
header = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.domain')

Veri Keşfi

Veri setini bulduğunuz web sayfasına bakarak bile çok fazla bilgi topladınız; ancak bu durumda Spark ve Python’un yardımıyla verinizi bizzat incelemek her zaman daha iyidir.

Burada anlaşılması gereken önemli nokta, Spark’ın yürütmesinin “tembel” olması nedeniyle henüz hiçbir şeyin yürütülmediğidir. Veriniz aslında içeri alınmadı. rdd ve header değişkenleri zihinlerdeki birer kavram gibidir. Spark’ı sizin için çalışmaya itmelisiniz; o hâlde header’a bakmak için collect() yöntemini kullanalım:

header.collect()

collect() yöntemi tüm RDD’yi tek bir makineye getirir ve aşağıdaki sonucu görürsünüz:

[u'longitude: continuous.', u'latitude: continuous.', u'housingMedianAge: continuous. ', u'totalRooms: continuous. ', u'totalBedrooms: continuous. ', u'population: continuous. ', u'households: continuous. ', u'medianIncome: continuous. ', u'medianHouseValue: continuous. ']

İpucu: collect() kullanırken dikkatli olun! Bu satırı çalıştırmak sürücünün belleğinin tükenmesine neden olabilir. Bu nedenle, RDD’nin yalnızca birkaç öğesini yazdırmak istiyorsanız aşağıdaki take() yöntemiyle yaklaşım daha güvenlidir. Genel olarak, sonuç kümenizi mümkün olduğunda sınırlamak iyi bir ilkedir; SQL kullanırken olduğu gibi.

Değişkenlerin sırasının, veri setinin sunumunda yukarıda gördüğünüzle aynı olduğunu ve tüm sütunların sürekli değerlere sahip olması gerektiğini öğrenirsiniz. Bunu doğrulamak için Spark’ı biraz daha çalıştırıp California konut verilerine göz atalım.

RDD’nizde take() yöntemini çağırın:

rdd.take(2)

Önceki satırı çalıştırarak RDD’nin ilk 2 öğesini alırsınız. Beklediğiniz gibi sonuç şudur: Dosyaları textFile() işleviyle okuduğunuz için satırların tümü birlikte okunmuştur. Girdiler tek bir virgülle ayrılır ve satırların kendileri de virgülle ayrılmıştır:

[u'-122.230000,37.880000,41.000000,880.000000,129.000000,322.000000,126.000000,8.325200,452600.000000', u'-122.220000,37.860000,21.000000,7099.000000,1106.000000,2401.000000,1138.000000,8.301400,358500.000000']

Bunu kesinlikle çözmeniz gerekir. Girdileri bölmenize gerek yok; ancak verinizin satırlarının ayrı öğeler olduğundan emin olmalısınız. Bunu çözmek için satırı virgülde bölen bir lambda işlevi geçirdiğiniz map() işlevini kullanacaksınız. Ardından, daha önce yaptığınız gibi take() yöntemiyle aynı satırı çalıştırarak sonucu kontrol edin:

Unutmayın lambda işlevleri çalışma zamanında oluşturulan anonim işlevlerdir.

# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)

Aşağıdaki sonucu alırsınız:

[[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000'], [u'-122.220000', u'37.860000', u'21.000000', u'7099.000000', u'1106.000000', u'2401.000000', u'1138.000000', u'8.301400', u'358500.000000']]

Alternatif olarak, verinizi incelemek için aşağıdaki işlevleri de kullanabilirsiniz:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)

Eğer Pandas veya R’deki data frame’lerle çalışmaya alışkınsanız, muhtemelen bir başlık görmeyi de beklersiniz; ancak yok. İşinizi kolaylaştırmak için RDD’den ilerleyip onu bir DataFrame’e dönüştüreceksiniz. Mümkün olduğunda DataFrame’ler RDD’lere tercih edilir. Özellikle Python ile çalışırken DataFrame’lerin performansı RDD’lerden daha iyidir.

Peki ikisi arasındaki fark nedir?

Yapılandırılmamış veriniz üzerinde düşük seviyeli dönüşümler ve eylemler yapmak istediğinizde RDD’leri kullanabilirsiniz. Bu, işleme sırasında bir şema dayatmayı veya özniteliklere ad veya sütuna göre erişmeyi önemsemediğiniz anlamına gelir. Daha önce performans hakkında söylenenlere bağlayacak olursak, RDD’leri kullanarak (yarı) yapılandırılmış veriler için DataFrame’lerin sunabileceği performans avantajlarına mutlaka ihtiyaç duymazsınız. Veriyi alan-özel ifadelerdense fonksiyonel programlama yapılarıyla değiştirmek istediğinizde RDD’leri kullanın.

Özetle, şimdi DataFrame’lere geçeceksiniz; böylece üst düzey ifadeler kullanabilir, verinizi daha fazla keşfetmek için SQL sorguları yapabilir ve sütun bazlı erişim elde edebilirsiniz.

Hadi başlayalım.

İlk adım bir SchemaRDD veya şemalı Row nesnelerinden oluşan bir RDD oluşturmaktır. Bu normaldir; çünkü bir DataFrame’de olduğu gibi sonunda satırlara ve sütunlara sahip olduğunuz bir duruma ulaşmak istersiniz. Her bir giriş bir satır ve belirli bir sütunla ilişkilidir ve sütunların veri tipleri vardır.

map() işlevini ve bir başka lambda işlevini tekrar kullanarak her bir girişi bir Row’daki alana eşleyeceksiniz. Bunu daha görsel hâle getirmek için şu ilk satırı düşünün:

[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000']

Lambda işlevi, bir SchemaRDD’de bir satır oluşturacağınızı ve 0. indisteki öğenin “longitude” adını alacağını vb. belirtir.

Bu SchemaRDD hazır olduğunda, toDF() yöntemiyle RDD’yi kolayca bir DataFrame’e dönüştürebilirsiniz.

# Import the necessary modules 
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()

Artık df DataFrame’inize sahip olduğunuza göre, daha önce kullandığınız first() ve take() yöntemlerinin yanı sıra head() ve show() ile de inceleyebilirsiniz:

# Show the top 20 rows 
df.show()

Hemen, bunun daha önce çalıştığınız RDD’den çok farklı göründüğünü fark edeceksiniz:

İpucu: DataFrame’inizin sütunlarını döndürmek için df.columns kullanın.

Veriler sütunlara güzelce yerleşmiş görünüyor; peki ya veri tipleri? Veriyi okurken Spark bir şema çıkarmaya çalışır; fakat burada başarılı oldu mu? DataFrame’inizdeki veri tipleri hakkında daha fazla bilgi edinmek için df.dtypes veya df.printSchema() kullanın.

# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()

İlk satırı çalıştırmadığınız için yalnızca aşağıdaki sonucu geri alırsınız:

root
 |-- households: string (nullable = true)
 |-- housingMedianAge: string (nullable = true)
 |-- latitude: string (nullable = true)
 |-- longitude: string (nullable = true)
 |-- medianHouseValue: string (nullable = true)
 |-- medianIncome: string (nullable = true)
 |-- population: string (nullable = true)
 |-- totalBedRooms: string (nullable = true)
 |-- totalRooms: string (nullable = true)

Tüm sütunlar hâlâ string veri tipinde… Bu hayal kırıklığı!

Bu DataFrame ile devam etmek istiyorsanız bu durumu düzeltmeli ve tüm sütunlara “daha iyi” veya daha doğru veri tipleri atamalısınız. Performansınız da bundan fayda görecektir. Sezgisel olarak, df DataFrame’inin her sütununun FloatType()’a döndürülmesi gerektiğini belirttiğiniz aşağıdaki gibi bir çözüm düşünebilirsiniz:

from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))

Ancak bu tekrarlanan çağrılar hem hataya açık, hem gözden kaçırması kolay hem de pek hoş görünmüyor. Neden tüm bunları daha temiz bir şekilde yapacak bir işlev yazmıyorsunuz?

Aşağıdaki Kullanıcı Tanımlı İşlev (UDF), bir DataFrame, sütun adları ve sütunlara vermek istediğiniz yeni veri tipini alır. Her sütun adı için o sütunu alıp yeni veri tipine döndürdüğünüzü söylersiniz. Sonra DataFrame’i döndürürsünüz:

# Import all from `sql.types`
from pyspark.sql.types import *

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
  for name in names: 
     df = df.withColumn(name, df[name].cast(newType))
  return df 

# Assign all column names to `columns`
columns = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome', 'population', 'totalBedRooms', 'totalRooms']

# Conver the `df` columns to `FloatType()`
df = convertColumn(df, columns, FloatType())

Bu çok daha iyi görünüyor! printSchema() yöntemiyle df’nin veri tiplerini hızlıca kontrol edebilirsiniz; tıpkı daha önce yaptığınız gibi.

Şimdi her şeyi yoluna koyduğunuza göre, gerçekten veri keşfine başlamanın zamanı geldi. Sütun bazlı erişim ve SQL sorgularının, DataFrame kullanmanın iki avantajı olduğunu görmüştünüz. Şimdi biraz daha derine inme zamanı. Küçük başlayalım ve df’den yalnızca 10 satırını görmek istediğiniz iki sütunu seçelim:

df.select('population','totalBedRooms').show(10)

Bu sorgu size aşağıdaki sonucu verir:

+----------+-------------+
|population|totalBedRooms|
+----------+-------------+
|     322.0|        129.0|
|    2401.0|       1106.0|
|     496.0|        190.0|
|     558.0|        235.0|
|     565.0|        280.0|
|     413.0|        213.0|
|    1094.0|        489.0|
|    1157.0|        687.0|
|    1206.0|        665.0|
|    1551.0|        707.0|
+----------+-------------+
only showing top 10 rows

Aşağıdaki örnekte gördüğünüz gibi sorgularınızı daha da karmaşık hâle getirebilirsiniz:

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()

Bu da size şu sonucu verir:

+----------------+-----+                                                        
|housingMedianAge|count|
+----------------+-----+
|            52.0| 1273|
|            51.0|   48|
|            50.0|  136|
|            49.0|  134|
|            48.0|  177|
|            47.0|  198|
|            46.0|  245|
|            45.0|  294|
|            44.0|  356|
|            43.0|  353|
|            42.0|  368|
|            41.0|  296|
|            40.0|  304|
|            39.0|  369|
|            38.0|  394|
|            37.0|  537|
|            36.0|  862|
|            35.0|  824|
|            34.0|  689|
|            33.0|  615|
+----------------+-----+
only showing top 20 rows

Sorgulamanın yanı sıra verinizi betimlemeyi ve bazı özet istatistikler almayı da seçebilirsiniz. Bu, sonrasında size kesinlikle yardımcı olur!

df.describe().show()

Tüm (sayısal) özniteliklerin minimum ve maksimum değerlerine bakın. Birden çok özniteliğin geniş bir değer aralığına sahip olduğunu görürsünüz: veri kümenizi normalize etmeniz gerekecek.

Veri Ön İşleme

Küçük keşifsel veri analizinizden topladığınız tüm bu bilgilerle, veriyi modele beslemek üzere ön işlemden geçirmek için yeterli bilgiye sahipsiniz.

• Eksik değerleri önemsememelisiniz; sıfır değerleri olan tüm girişler veri setinden çıkarılmıştır.
• Minimum ve maksimum değerlerin aralığının oldukça büyük olduğunu gördüğünüz için verinizi muhtemelen standartlaştırmalısınız.
• Oda başına yatak odası sayısını veya hane başına oda sayısını kaydeden bir özellik gibi eklenebilecek bazı ilave öznitelikler olabilir.
• Bağımlı değişkeniniz de oldukça büyük; işinizi kolaylaştırmak için değerleri biraz ayarlamanız gerekecek.

# Import all from `sql.functions` 
from pyspark.sql.functions import *

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

# Show the first 2 lines of `df`
df.take(2)

[Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0), Row(households=1138.0, housingMedianAge=21.0, latitude=37.86000061035156, longitude=-122.22000122070312, medianHouseValue=3.585, medianIncome=8.301400184631348, population=2401.0, totalBedRooms=1106.0, totalRooms=7099.0)]

# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()

Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0, roomsPerHousehold=6.984126984126984, populationPerHousehold=2.5555555555555554, bedroomsPerRoom=0.14659090909090908)

# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")

# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])

# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)

[Row(label=4.526, features=DenseVector([129.0, 322.0, 126.0, 8.3252, 6.9841, 2.5556, 0.1466]), features_scaled=DenseVector([0.3062, 0.2843, 0.3296, 4.3821, 2.8228, 0.2461, 2.5264])), Row(label=3.585, features=DenseVector([1106.0, 2401.0, 1138.0, 8.3014, 6.2381, 2.1098, 0.1558]), features_scaled=DenseVector([2.6255, 2.1202, 2.9765, 4.3696, 2.5213, 0.2031, 2.6851]))]

# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)

# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]

[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]

# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.intercept

# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462

# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2

# RMSE
0.8692118678997669

# R2
0.4240895287218379

spark.stop()