Tutorial Apache Spark: ML dengan PySpark

Mengunduh pyspark dengan pip

Kemudian, Anda dapat mengunduh dan memasang PySpark dengan bantuan pip. Ini cukup mudah dan mirip dengan memasang paket lainnya. Anda cukup menjalankan perintah biasa dan pekerjaan beratnya akan dilakukan untuk Anda:

$ pip install pyspark

Sebagai alternatif, Anda juga dapat pergi ke halaman unduhan Spark. Biarkan opsi default di tiga langkah pertama dan Anda akan menemukan tautan unduhan di langkah ke-4. Klik tautan tersebut untuk mengunduhnya. Untuk tutorial ini, Anda akan mengunduh Spark Release 2.2.0 dan tipe paket “Pre-built for Apache Hadoop 2.7 and later”.

Catatan bahwa unduhan mungkin memerlukan waktu untuk selesai!

Mengunduh Spark dengan Homebrew

Anda juga dapat memasang Spark dengan Homebrew, pengelola paket gratis dan open source. Ini sangat membantu jika Anda bekerja dengan macOS.

Cukup jalankan perintah berikut untuk mencari Spark, mendapatkan lebih banyak informasi, dan akhirnya memasangnya di komputer pribadi Anda:

# Search for spark
$ brew search spark

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

# Install apache-spark
$ brew install apache-spark

Unduh dan Siapkan Spark

Selanjutnya, pastikan Anda mengekstrak (untar) direktori yang muncul di folder Downloads Anda. Ini bisa terjadi otomatis untuk Anda dengan mengklik ganda arsip spark-2.2.0-bin-hadoop2.7.tgz atau dengan membuka Terminal dan menjalankan perintah berikut:

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

Kemudian, pindahkan folder hasil ekstrak ke /usr/local/spark dengan menjalankan baris berikut:

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

Catatan jika Anda mendapat galat yang mengatakan izin ditolak untuk memindahkan folder ini ke lokasi baru, Anda harus menambahkan sudo di depan perintah ini. Baris di atas akan menjadi $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Anda akan diminta memasukkan kata sandi, yang biasanya sama dengan yang Anda gunakan untuk membuka PC saat menyalakannya :)

Sekarang semuanya siap, buka berkas README di jalur /usr/local/spark. Anda dapat melakukannya dengan menjalankan

$ cd /usr/local/spark

Ini akan membawa Anda ke folder yang dibutuhkan. Lalu, Anda dapat mulai menelusuri folder dan membaca berkas README yang disertakan di dalamnya.

Pertama, gunakan $ ls untuk mendapatkan daftar berkas dan folder yang ada di folder spark ini. Anda akan melihat ada berkas README.md di sana. Anda dapat membukanya dengan menjalankan salah satu perintah berikut:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

Tip gunakan tombol tab pada keyboard Anda untuk melengkapi otomatis saat mengetik nama berkas :) Ini akan menghemat waktu Anda.

Anda akan melihat bahwa README ini memberi Anda beberapa informasi umum tentang Spark, dokumentasi daring, membangun Spark, shell interaktif Scala dan Python, contoh program, dan banyak lagi.

Hal yang mungkin paling menarik di sini adalah bagian tentang cara membangun Spark, tetapi perhatikan bahwa ini hanya relevan jika Anda tidak mengunduh versi pre-built. Untuk tutorial ini, Anda telah mengunduh versi pre-built. Anda dapat menekan CTRL + X untuk keluar dari README, yang akan membawa Anda kembali ke folder spark.

Jika Anda memilih versi yang belum dibangun, pastikan Anda menjalankan perintah yang tercantum di berkas README. Pada saat penulisan, perintahnya sebagai berikut:

$ build/mvn -DskipTests clean package run

Perhatikan bahwa perintah ini dapat memakan waktu untuk dijalankan.

Aplikasi Spark vs Spark Shell

Shell interaktif adalah contoh lingkungan Read-Eval(uate)-Print-Loop (REPL); Artinya, apa pun yang Anda ketik akan dibaca, dievaluasi, dan dicetak kembali kepada Anda sehingga Anda dapat melanjutkan analisis. Ini mungkin mengingatkan Anda pada IPython, shell Python interaktif yang kuat yang mungkin Anda kenal dari bekerja dengan Jupyter. Jika ingin tahu lebih banyak, pertimbangkan membaca posting blog DataCamp IPython or Jupyter.

Ini berarti Anda dapat menggunakan shell, yang tersedia untuk Python maupun Scala, untuk semua pekerjaan interaktif yang perlu Anda lakukan.

Selain shell ini, Anda juga dapat menulis dan menerapkan aplikasi Spark. Berbeda dengan menulis aplikasi Spark, SparkSession sudah dibuat untuk Anda sehingga Anda bisa langsung bekerja dan tidak membuang waktu berharga untuk membuatnya.

Sekarang Anda mungkin bertanya-tanya: apa itu SparkSession?

Nah, itu adalah titik masuk utama untuk fungsionalitas Spark: ia merepresentasikan koneksi ke cluster Spark dan Anda dapat menggunakannya untuk membuat RDD dan menyiarkan variabel di cluster tersebut. Saat Anda bekerja dengan Spark, semuanya dimulai dan diakhiri dengan SparkSession ini. Catatan bahwa sebelum Spark 2.0.0, tiga objek koneksi utama adalah SparkContext, SqlContext, dan HiveContext.

Anda akan melihat lebih lanjut tentang ini nanti. Untuk saat ini, mari fokus pada shell.

Python Spark Shell

Dari dalam folder spark yang terletak di /usr/local/spark, Anda dapat menjalankan

$ ./bin/pyspark

Pertama, Anda akan melihat beberapa teks muncul. Lalu, Anda akan melihat “Spark” muncul, seperti ini:

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'.
>>>

Saat Anda melihat ini, Anda tahu bahwa Anda siap mulai bereksperimen di dalam shell interaktif!

Tip: jika Anda lebih suka menggunakan shell IPython daripada Spark shell, Anda dapat melakukannya dengan menyetel variabel lingkungan berikut:

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

PySpark di Jupyter Notebook

Untuk bagian tutorial ini, Anda tidak akan menggunakan ishell melainkan membangun aplikasi Anda sendiri. Anda akan melakukannya di Jupyter Notebook. Anda sudah memiliki semua yang diperlukan terpasang, jadi Anda tidak perlu banyak hal untuk membuat PySpark bekerja di Jupyter.

Anda dapat meluncurkan aplikasi notebook seperti biasa, dengan menjalankan $ jupyter notebook. Lalu, buat notebook baru dan cukup impor pustaka findspark dan gunakan fungsi init(). Dalam kasus ini, Anda akan memberikan path /usr/local/spark ke init() karena Anda yakin inilah path tempat Anda memasang Spark.

# Import findspark 
import findspark

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

# Or use this alternative
#findspark.init()

Tip: jika Anda tidak tahu apakah path sudah disetel dengan benar atau di mana Anda memasang Spark di PC, Anda selalu dapat menggunakan findspark.find() untuk mendeteksi lokasi pemasangan Spark secara otomatis.

Jika Anda mencari cara alternatif untuk bekerja dengan Spark di Jupyter, lihat Apache Spark in Python: Beginner’s Guide kami.

Sekarang semua sudah beres, Anda akhirnya dapat mulai membuat program Spark pertama Anda!

Membuat Program Spark Pertama Anda

Hal pertama yang ingin Anda lakukan adalah mengimpor SparkContext dari paket pyspark dan menginisialisasinya. Ingat bahwa Anda tidak perlu melakukan ini sebelumnya karena shell interaktif Spark otomatis membuat dan menginisialisasinya untuk Anda! Di sini, Anda perlu melakukan sedikit lebih banyak pekerjaan sendiri :)

Impor modul SparkSession dari pyspark.sql dan bangun SparkSession dengan metode builder(). Setelah itu, Anda dapat menyetel URL master untuk terhubung, nama aplikasi, menambahkan beberapa konfigurasi tambahan seperti memori executor, dan terakhir gunakan getOrCreate() untuk mengambil sesi Spark saat ini atau membuat satu jika belum ada yang berjalan.

# 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

Catatan jika Anda mendapat galat FileNotFoundError serupa ini: “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”, Anda tahu bahwa Anda harus (meng)atur ulang PATH Spark Anda. Pergi ke direktori home dengan menjalankan $ cd lalu edit berkas .bash_profile dengan menjalankan $ nano .bash_profile.

Tambahkan sesuatu seperti berikut di bagian bawah berkas

export SPARK_HOME="/usr/local/spark"

Gunakan CTRL + X untuk keluar dari berkas namun pastikan menyimpan penyesuaian dengan menekan Y untuk mengonfirmasi perubahan. Selanjutnya, jangan lupa menerapkan perubahan dengan menjalankan source .bash_profile.

Tip: Anda juga dapat menyetel variabel lingkungan tambahan jika diinginkan; Mungkin Anda tidak membutuhkannya, tetapi sebaiknya mengetahui bahwa Anda dapat menyetelnya jika diperlukan. Pertimbangkan contoh berikut:

# 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/

Catatan juga bahwa sekarang Anda telah menginisialisasi SparkSession default. Namun, dalam banyak kasus, Anda akan ingin mengonfigurasinya lebih lanjut. Anda akan melihat ini sangat dibutuhkan saat mulai bekerja dengan big data. Jika ingin tahu lebih lanjut, cek halaman ini.

Memuat Data Anda

Tutorial ini menggunakan himpunan data California Housing. Data ini muncul dalam makalah tahun 1997 berjudul Sparse Spatial Autoregressions, ditulis oleh Pace, R. Kelley dan Ronald Barry dan diterbitkan di jurnal Statistics and Probability Letters. Para peneliti membangun himpunan data ini menggunakan data sensus California tahun 1990.

Data berisi satu baris per kelompok blok sensus. Kelompok blok adalah unit geografis terkecil yang digunakan Biro Sensus A.S. untuk menerbitkan data sampel (satu kelompok blok biasanya memiliki populasi 600 hingga 3.000 orang). Dalam sampel ini, satu kelompok blok rata-rata mencakup 1425,5 individu yang tinggal di area geografis yang kompak. Anda akan mendapatkan informasi ini dari laman web ini atau dengan membaca makalah yang disebutkan di atas dan dapat Anda temukan di sini.

Data spasial ini berisi 20.640 observasi tentang harga perumahan dengan 9 variabel ekonomi:

• Longitude mengacu pada jarak sudut suatu tempat geografis ke utara atau selatan khatulistiwa bumi untuk setiap kelompok blok;
• Latitude mengacu pada jarak sudut suatu tempat geografis ke timur atau barat khatulistiwa bumi untuk setiap kelompok blok;
• Housing median age adalah usia median orang yang termasuk dalam suatu kelompok blok. Catatan bahwa median adalah nilai yang berada di titik tengah distribusi frekuensi dari nilai yang diamati;
• Total rooms adalah jumlah total kamar di rumah per kelompok blok;
• Total bedrooms adalah jumlah total kamar tidur di rumah per kelompok blok;
• Population adalah jumlah penduduk dari suatu kelompok blok;
• Households mengacu pada unit rumah dan penghuninya per kelompok blok;
• Median income digunakan untuk mencatat pendapatan median orang yang termasuk dalam suatu kelompok blok; Dan,
• Median house value adalah variabel dependen dan mengacu pada nilai rumah median per kelompok blok.

Selain itu, Anda juga mengetahui bahwa semua kelompok blok yang memiliki entri nol untuk variabel independen dan dependen telah dikeluarkan dari data.

Median house value adalah variabel dependen dan akan ditetapkan sebagai variabel target dalam model ML Anda.

Anda dapat mengunduh datanya di sini. Cari folder houses.zip, unduh dan ekstrak sehingga Anda dapat mengakses folder data.

Selanjutnya, Anda akan menggunakan metode textFile() untuk membaca data dari folder tempat Anda mengunduhnya ke RDD. Metode ini menerima URI untuk berkas, yang dalam kasus ini adalah path lokal mesin Anda, dan membacanya sebagai kumpulan baris. Demi kemudahan, Anda tidak hanya akan membaca berkas .data, tetapi juga berkas .domain yang berisi header. Ini akan memungkinkan Anda mengecek ulang urutan variabel.

# 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')
Eksplorasi Data
Anda sudah mengumpulkan banyak informasi hanya dengan melihat laman web tempat Anda menemukan himpunan data, tetapi selalu lebih baik untuk terjun langsung dan memeriksa data Anda dengan bantuan Spark dengan Python, dalam hal ini.
Hal penting untuk dipahami di sini adalah, karena eksekusi Spark bersifat “malas” (lazy), belum ada yang dieksekusi. Data Anda sebenarnya belum dibaca. Variabel rdd dan header sebenarnya hanyalah konsep di benak Anda. Anda harus mendorong Spark untuk bekerja untuk Anda, jadi mari gunakan metode collect() untuk melihat header:
header.collect()
Metode collect() membawa seluruh RDD ke satu mesin, dan Anda akan melihat hasil berikut:
[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. ']
Tip: berhati-hatilah saat menggunakan collect()! Menjalankan baris kode ini dapat menyebabkan driver kehabisan memori. Itulah mengapa pendekatan berikut dengan metode take() lebih aman jika Anda hanya ingin mencetak beberapa elemen RDD. Secara umum, merupakan prinsip yang baik untuk membatasi himpunan hasil kapan pun memungkinkan, seperti saat Anda menggunakan SQL.
Anda mengetahui bahwa urutan variabel sama dengan yang Anda lihat di presentasi himpunan data di atas, dan Anda juga mengetahui bahwa semua kolom seharusnya memiliki nilai kontinu. Mari paksa Spark bekerja lebih keras dan lihat data California housing untuk mengonfirmasinya.
Panggil metode take() pada RDD Anda:
rdd.take(2)
Dengan menjalankan baris kode sebelumnya, Anda mengambil 2 elemen pertama dari RDD. Hasilnya seperti yang diharapkan: karena Anda membaca berkas dengan fungsi textFile(), semua baris dibaca bersama-sama. Entri dipisahkan oleh koma tunggal dan barisnya sendiri juga dipisahkan oleh koma:
[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']
Anda jelas perlu menyelesaikan ini. Sekarang, Anda tidak perlu memisah entri, tetapi Anda harus memastikan bahwa baris data Anda adalah elemen yang terpisah. Untuk menyelesaikannya, Anda akan menggunakan fungsi map() yang Anda beri fungsi lambda untuk memisah baris pada koma. Lalu, periksa hasilnya dengan menjalankan baris yang sama dengan metode take(), seperti yang Anda lakukan sebelumnya:
Ingat bahwa fungsi lambda adalah fungsi anonim yang dibuat saat runtime.
# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)
Anda akan mendapatkan hasil berikut:
[[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']]
Sebagai alternatif, Anda juga dapat menggunakan fungsi berikut untuk memeriksa data:
# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)
Jika Anda terbiasa bekerja dengan Pandas atau data frame di R, Anda mungkin juga mengharapkan melihat header, tetapi tidak ada. Untuk mempermudah, Anda akan beralih dari RDD dan mengonversinya ke DataFrame. DataFrame lebih disukai daripada RDD kapan pun memungkinkan. Terutama saat Anda bekerja dengan Python, performa DataFrame lebih baik daripada RDD.
Lalu apa perbedaan keduanya?
Anda dapat menggunakan RDD saat ingin melakukan transformasi dan aksi tingkat rendah pada data tidak terstruktur Anda. Ini berarti Anda tidak peduli untuk memberlakukan skema saat memproses atau mengakses atribut berdasarkan nama atau kolom. Terkait dengan apa yang dikatakan sebelumnya tentang performa, dengan menggunakan RDD, Anda tidak selalu menginginkan manfaat performa yang dapat ditawarkan DataFrame untuk data (semi) terstruktur. Gunakan RDD saat Anda ingin memanipulasi data dengan konstruksi pemrograman fungsional daripada ekspresi spesifik domain.
Untuk merangkum, Anda akan beralih ke DataFrame sekarang untuk menggunakan ekspresi tingkat tinggi, melakukan kueri SQL untuk mengeksplorasi data lebih lanjut, dan mendapatkan akses kolumnar.
Jadi mari kita lakukan.
Langkah pertama adalah membuat SchemaRDD atau RDD dari objek Row dengan skema. Ini wajar, karena seperti DataFrame, pada akhirnya Anda ingin berada pada situasi di mana Anda memiliki baris dan kolom. Setiap entri ditautkan ke baris dan kolom tertentu dan kolom memiliki tipe data.
Anda akan menggunakan fungsi map() lagi dan fungsi lambda lain di mana Anda akan memetakan setiap entri ke sebuah field dalam Row. Untuk memvisualisasikannya, perhatikan baris pertama ini:
[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']
Fungsi lambda menyatakan bahwa Anda akan menyusun sebuah baris dalam SchemaRDD dan bahwa elemen pada indeks 0 akan memiliki nama “longitude”, dan seterusnya.
Dengan SchemaRDD ini, Anda dapat dengan mudah mengonversi RDD ke DataFrame dengan metode toDF().
# 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()
Sekarang Anda memiliki DataFrame df, Anda dapat memeriksanya dengan metode yang juga Anda gunakan sebelumnya, yaitu first() dan take(), tetapi juga dengan head() dan show():
# Show the top 20 rows 
df.show()
Anda akan segera melihat bahwa ini terlihat sangat berbeda dari RDD yang Anda kerjakan sebelumnya:



Tip: gunakan df.columns untuk mengembalikan kolom-kolom DataFrame Anda.
Data tampak tersusun rapi ke dalam kolom, tetapi bagaimana dengan tipe datanya? Dengan membaca data, Spark akan mencoba menyimpulkan skema, namun apakah ini berhasil di sini? Gunakan df.dtypes atau df.printSchema() untuk mengetahui lebih lanjut tentang tipe data yang terdapat dalam DataFrame Anda.
# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()
Karena Anda tidak mengeksekusi baris kode pertama, Anda hanya akan mendapatkan hasil berikut:
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)
Semua kolom masih bertipe data string… Itu mengecewakan!
Jika Anda ingin melanjutkan dengan DataFrame ini, Anda perlu memperbaiki situasi ini dan menetapkan tipe data yang “lebih baik” atau lebih akurat untuk semua kolom. Performa Anda juga akan diuntungkan. Secara intuitif, Anda bisa memilih solusi seperti berikut, di mana Anda menyatakan bahwa setiap kolom dari DataFrame df harus di-cast ke FloatType():
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()))
Namun pemanggilan berulang ini cukup tidak rapi, rawan salah, dan tidak enak dilihat. Mengapa tidak menulis fungsi yang dapat melakukan semua ini untuk Anda dengan cara yang lebih bersih?
User-Defined Function (UDF) berikut menerima DataFrame, nama kolom, dan tipe data baru yang Anda inginkan untuk kolom-kolom tersebut. Anda menyatakan bahwa untuk setiap nama kolom, Anda mengambil kolom itu dan me-cast-nya ke tipe data baru. Lalu, Anda mengembalikan DataFrame:
# 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())
Itu sudah terlihat jauh lebih baik! Anda dapat dengan cepat memeriksa tipe data df dengan metode printSchema(), seperti yang telah Anda lakukan sebelumnya.
Sekarang setelah semua itu beres, saatnya benar-benar memulai eksplorasi data. Anda telah melihat bahwa akses kolumnar dan kueri SQL adalah dua keuntungan menggunakan DataFrame. Nah, sekarang saatnya menggali lebih jauh. Mari mulai kecil dan pilih dua kolom dari df yang hanya ingin Anda lihat 10 barisnya:
df.select('population','totalBedRooms').show(10)
Kueri ini memberi Anda hasil berikut:
+----------+-------------+
|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
Anda juga dapat membuat kueri Anda lebih kompleks, seperti pada contoh berikut:
df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()
Yang memberi Anda hasil berikut:
+----------------+-----+                                                        
|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
Selain melakukan kueri, Anda juga dapat memilih untuk mendeskripsikan data Anda dan mendapatkan statistik ringkasan. Ini hampir pasti akan membantu Anda nanti!
df.describe().show()





Lihat nilai minimum dan maksimum dari semua atribut (numerik). Anda melihat bahwa beberapa atribut memiliki rentang nilai yang lebar: Anda akan perlu menormalkan dataset Anda.
Prapemrosesan Data
Dengan semua informasi yang Anda kumpulkan dari analisis eksploratif kecil Anda, Anda cukup tahu untuk melakukan prapemrosesan data agar dapat diberi ke model.

Anda tidak perlu khawatir tentang nilai hilang; semua nilai nol telah dikeluarkan dari himpunan data.
Anda sebaiknya melakukan standardisasi data, karena Anda melihat bahwa rentang nilai minimum dan maksimum cukup besar.
Mungkin ada beberapa atribut tambahan yang dapat Anda tambahkan, seperti fitur yang mencatat jumlah kamar tidur per kamar atau kamar per rumah tangga.
Variabel dependen Anda juga cukup besar; Untuk mempermudah, Anda harus sedikit menyesuaikan nilainya.


Prapemrosesan Nilai Target
Pertama, mari mulai dengan medianHouseValue, variabel dependen Anda. Untuk memudahkan Anda bekerja dengan nilai target, Anda akan mengekspresikan nilai rumah dalam satuan 100.000. Itu berarti target seperti 452600.000000 harus menjadi 4.526:
# 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)
Anda dapat dengan jelas melihat bahwa nilainya telah disesuaikan dengan benar saat melihat hasil metode take():
[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)]


Rekayasa Fitur
Sekarang setelah Anda menyesuaikan nilai dalam medianHouseValue, Anda juga dapat menambahkan variabel tambahan yang Anda baca di atas. Anda akan menambahkan kolom-kolom berikut ke himpunan data:

Rooms per household yang mengacu pada jumlah kamar dalam rumah tangga per kelompok blok;
Population per household, yang pada dasarnya memberi indikasi berapa banyak orang yang tinggal dalam rumah tangga per kelompok blok; Dan
Bedrooms per room yang akan memberi gambaran berapa banyak kamar yang merupakan kamar tidur per kelompok blok;



Karena Anda bekerja dengan DataFrame, Anda paling baik menggunakan metode select() untuk memilih kolom yang akan dikerjakan, yaitu totalRooms, households, dan population. Selain itu, Anda harus menunjukkan bahwa Anda bekerja dengan kolom dengan menambahkan fungsi col() ke kode Anda. Jika tidak, Anda tidak akan dapat melakukan operasi elemen per elemen seperti pembagian yang Anda pikirkan untuk ketiga variabel ini:
# 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()
Anda melihat bahwa, untuk baris pertama, ada sekitar 6,98 kamar per rumah tangga, rumah tangga dalam kelompok blok terdiri dari sekitar 2,5 orang dan jumlah kamar tidur cukup rendah yaitu 0,14:
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)
Berikutnya, —dan ini sudah mengantisipasi masalah yang mungkin Anda temui saat menstandardisasi nilai dalam himpunan data— Anda juga akan mengurutkan ulang nilai. Karena Anda tidak ingin menstandardisasi nilai target, Anda ingin memastikan untuk mengisolasinya dalam himpunan data Anda.
Dalam hal ini, Anda perlu melakukannya dengan menggunakan metode select() dan meneruskan nama kolom dalam urutan yang lebih tepat. Dalam kasus ini, variabel target medianHouseValue diletakkan di awal, sehingga tidak akan terpengaruh oleh standardisasi.
Catatan juga bahwa ini adalah waktu yang tepat untuk mengeluarkan variabel yang mungkin tidak ingin Anda pertimbangkan dalam analisis. Dalam kasus ini, mari keluarkan variabel seperti longitude, latitude, housingMedianAge, dan totalRooms.
# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")


Standardisasi
Sekarang setelah Anda mengurutkan ulang data, Anda siap menormalkan data. Atau hampir siap. Masih ada satu langkah lagi yang perlu Anda lakukan: memisahkan fitur dari variabel target. Intinya, ini berarti mengisolasi kolom pertama di DataFrame Anda dari kolom lainnya.
Dalam hal ini, Anda akan menggunakan fungsi map() yang Anda gunakan dengan RDD untuk melakukan aksi ini. Anda juga melihat bahwa Anda menggunakan fungsi DenseVector(). Dense vector adalah vektor lokal yang didukung oleh array double yang merepresentasikan nilai entri. Dengan kata lain, ini digunakan untuk menyimpan array nilai untuk digunakan di PySpark.
Selanjutnya, Anda kembali membuat DataFrame dari input_data dan memberi label ulang kolom dengan meneruskan sebuah list sebagai argumen kedua. List ini terdiri dari nama kolom "label" dan "features":
# 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"])
Selanjutnya, Anda akhirnya dapat men-skala data. Anda dapat menggunakan Spark ML untuk melakukan ini: pustaka ini akan membuat machine learning pada big data menjadi skalabel dan mudah. Anda akan menemukan alat seperti algoritma ML dan semua yang Anda butuhkan untuk membangun pipeline ML praktis. Dalam kasus ini, Anda tidak perlu melakukan terlalu banyak prapemrosesan sehingga pipeline mungkin berlebihan, tetapi jika ingin mempelajarinya, pertimbangkan untuk mengunjungi halaman ini.
Kolom masukan adalah fitur, dan kolom keluaran dengan hasil yang telah diskalakan yang akan disertakan dalam scaled_df akan diberi nama "features_scaled":
# 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)
Mari lihat DataFrame Anda dan hasilnya. Anda melihat bahwa, memang, kolom ketiga features_scaled ditambahkan ke DataFrame Anda, yang dapat Anda gunakan untuk membandingkan dengan features:
[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]))]
Catatan bahwa baris-baris kode ini sangat mirip dengan yang akan Anda lakukan di Scikit-Learn.
Membangun Model Machine Learning dengan Spark ML
Dengan semua prapemrosesan selesai, akhirnya saatnya mulai membangun model Regresi Linear Anda! Seperti biasa, pertama-tama Anda perlu membagi data menjadi himpunan latih dan uji. Untungnya, ini bukan masalah dengan metode randomSplit():
# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)
Anda meneruskan sebuah list dengan dua angka yang mewakili ukuran yang Anda inginkan untuk himpunan latih dan uji serta seed, yang diperlukan untuk alasan reprodusibilitas. Jika ingin tahu lebih banyak tentang ini, pertimbangkan Python Machine Learning Tutorial dari DataCamp.
Lalu, tanpa basa-basi, Anda dapat membuat model Anda!
Catatan bahwa argumen elasticNetParam berkaitan dengan α atau intersep vertikal dan bahwa regParam atau parameter regularisasi berkaitan dengan λ. Kunjungi halaman ini untuk informasi lebih lanjut.
# 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)
Dengan model Anda siap, Anda dapat menghasilkan prediksi untuk data uji Anda: gunakan metode transform() untuk memprediksi label untuk test_data Anda. Lalu, Anda dapat menggunakan operasi RDD untuk mengekstrak prediksi serta label sebenarnya dari DataFrame dan memasangkan kedua nilai ini bersama-sama dalam daftar bernama predictionAndLabel.
Terakhir, Anda kemudian dapat memeriksa nilai prediksi dan nyata dengan cukup mengakses daftar dengan tanda kurung siku []:
# 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]
Anda akan melihat nilai nyata dan nilai prediksi berikut (dalam urutan tersebut):
[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]
Mengevaluasi Model
Melihat nilai prediksi adalah satu hal, tetapi hal lain yang lebih baik adalah melihat beberapa metrik untuk mendapatkan gambaran yang lebih baik tentang seberapa baik model Anda sebenarnya. Anda dapat memulai dengan mencetak koefisien dan intersep model:
# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.intercept
Yang memberi Anda hasil berikut:
# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462
Selanjutnya, Anda juga dapat menggunakan atribut summary untuk menampilkan rootMeanSquaredError dan r2:
# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2


RMSE mengukur seberapa besar kesalahan antara dua himpunan data dengan membandingkan nilai prediksi dan nilai yang diamati atau diketahui. Semakin kecil nilai RMSE, semakin dekat nilai prediksi dan nilai yang diamati.


R2 (“R squared”) atau koefisien determinasi adalah ukuran yang menunjukkan seberapa dekat data dengan garis regresi yang dipasang. Skor ini selalu antara 0 dan 100% (atau 0 hingga 1 dalam kasus ini), di mana 0% menunjukkan bahwa model tidak menjelaskan variabilitas data respons di sekitar rata-ratanya, dan 100% menunjukkan sebaliknya: model menjelaskan semua variabilitas. Artinya, secara umum, semakin tinggi R-squared, semakin baik model menyesuaikan data Anda.


Anda akan mendapatkan hasil berikut:
# RMSE
0.8692118678997669

# R2
0.4240895287218379



Model Anda jelas masih perlu ditingkatkan! Jika Anda ingin melanjutkan dengan model ini, Anda dapat bermain-main dengan parameter yang Anda berikan ke model, variabel yang Anda sertakan di DataFrame asli, .... Namun di sinilah tutorial ini berakhir untuk saat ini! 
Sebelum Anda Pergi…
Sebelum Anda pergi, pastikan untuk menghentikan SparkSession dengan baris kode berikut:
spark.stop()


Mengembangkan Big Data Lebih Jauh
Selamat! Anda telah sampai di akhir tutorial ini, di mana Anda mempelajari cara membuat model regresi linear dengan bantuan Spark ML.
Jika Anda tertarik mempelajari lebih lanjut tentang PySpark, pertimbangkan untuk mengikuti kursus Introduction to PySpark dari DataCamp dan lihat Tutorial Apache Spark: ML with PySpark.