Belajar PySpark - GroupBy dan Agregasi

Dalam artikel ini kita akan mempelajari mengenai groupBy dan agg, yaitu fungsi-fungsi untuk melakukan grouping dan agregasi pada pyspark DataFrame.

Bagi anda yang familiar dengan SQL atau pandas, groupBy pada pyspark dataframe serupa dengan klausa GROUP BY pada SQL ataupun pada pandas dataframe. Fungsinya adalah untuk mengelompokkan data berdasar satu atau beberapa nilai kolom yang sama.

Berikut ini dataframe yang akan kita gunakan dalam latihan:

data = [['Agus','Fisika',130,100],['Windy','Fisika',200,150],
        ['Budi','Biologi',200,150],['Dina','Fisika',150,200],
        ['Bayu','Fisika',150,200],['Dedi','Biologi',50,150]]

kolom = ["nama","jurusan","nilai1","nilai2"]

df = spark.createDataFrame(data,kolom)
df.show()

dataframe input

Grouping data dengan groupBy

Fungsi groupBy mengembalikan object bertipe pyspark.sql.group.GroupedData. Object ini memiliki beberapa fungsi yang diperlukan untuk melakukan agregasi, yaitu : sum, avg, max, min, count, dan agg.

result = df.groupBy("jurusan")
type(result)

Agregasi dengan sum, avg, min dan max

Jika tidak ditentukan kolom mana yang akan diagregasi (tanpa parameter), maka akan dilakukan fungsi agregasi yang sama untuk semua kolom numerik

result.avg().show()

agregasi pyspark dataframe dengan avg

Kita bisa memilih satu atau beberapa kolom yang akan diagregasi dengan menuliskan nama kolom sebagai parameter

df.groupBy("jurusan","semester").avg("nilai1").show()

memilih kolom agregasi

Dengan beberapa kolom yang di-agregasi

df.groupBy("jurusan","semester").min("nilai1","nilai2").show()

memilih beberapa kolom agregasi

Agregasi dengan count

Berbeda dengan fungsi lainnya, count() tidak menerima parameter apapun. Fungsi count() mengembalikan jumlah record tiap group

df.groupBy("jurusan").count().show()

agregasi dataframe dengan count

Agregasi dengan fungsi agg()

Fungsi agg menerima parameter berupa fungsi-fungsi agregasi dari pyspark.sql.functions atau dictionary mapping {namakolom : fungsi agregasi}.

Kelebihan menggunakan agg() diantaranya adalah :

pilihan fungsi agregasi yang lebih banyak,
dapat menggunakan beberapa fungsi agregasi yang berbeda dalam satu grouping, dan
dapat menggunakan alias untuk menamai kolom hasil agregasi.

Contoh agg() dengan satu fungsi agregasi

df.groupBy("jurusan").agg(F.avg("nilai2")).show()

agg dengan fungsi agregasi

Memanggil beberapa fungsi agregasi

df.groupBy("jurusan").agg(F.min("nilai1"),
                          F.max("nilai1")).show()

beberapa fungsi agregasi

Menggunakan alias

Untuk memudahkan akses DataFrame hasil agregasi, kita bisa memberi nama kolom output dengan menggunakan fungsi alias()

df.groupBy("jurusan") \
    .agg(F.min("nilai1").alias("min_nilai1"),
         F.max("nilai1").alias("max_nilai1"),
         F.median("nilai1").alias("median_nilai1"),
         F.count("nilai1").alias("jumlah_siswa")).show()

alias pada agregasi dataframe

Agregasi dengan agg dan dictionary

Kita dapat juga menggunakan parameter berupa dictionary. Akan tetapi cara ini cukup jarang digunakan karena sulit dibaca dan kurang fleksibel.

df.groupBy().agg(
    {"*": "count","nilai1":"min","nilai2":"max"}).show()

agg dengan parameter dictionary

Agregasi dengan fungsi expr()

Selain menggunakan fungsi-fungsi agregasi, kita juga dapat menggunakan fungsi expr(). Fungsi ini bermanfaat jika kita ingin melakukan pemrosesan kolom sebelum diagregasi. Misalnya seperti contoh di bawah ini

df.groupBy("jurusan", "semester").\
    agg(F.expr("avg(round((nilai1 + nilai2)/5)) as rerata_total")).show()

agregasi dengan agg() dan expr()

File jupyter notebook untuk latihan ini dapat diunduh di sini.

Artikel sebelumnya :