Belajar PySpark - ArrayType pada DataFrame Schema

Dalam PySpark, StructType adalah tipe data yang digunakan untuk mendefinisikan skema (schema) dari suatu DataFrame. Data yang disimpan dalam skema dapat berupa data sederhana, seperti integer atau string, maupun data kompleks, seperti map atau array.

Dalam artikel ini, kita akan membahas salah satu tipe data kompleks yang dapat disimpan dalam Struct Type, yaitu array.

Sebelumnya, seperti biasa kita import package yang akan digunakan dan kita buat dahulu spark session.

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, ArrayType, MapType, StringType, IntegerType
from pyspark.sql.functions import array_contains, explode, split, array

spark = SparkSession.builder.appName("Belajar PySpark - ArrayType").getOrCreate()

Membuat ArrayType Pada PySpark

ArrayType adalah data kompleks yang digunakan untuk menyimpan data dalam bentuk array atau list. Data dalam array sendiri dapat berupa data sederhana maupun data kompleks.

Sintaks untuk mendefinisikan sebuah field dengan tipe array adalah

StructField(“namakolom”, ArrayType(elementType = pyspark.sql.types.DataType).

elementType dapat berupa simple type seperti StringType, IntegerType, dan lain sebagainya, maupun kompleks type seperti StructType, MapType, atau ArrayType sendiri.

Contoh definisi ArrayType dalam skema DataFrame :

mySchema = StructType([
    StructField("nama", StringType(), True),
    StructField("jurusan", StringType(), True),
    StructField("nilai", ArrayType(elementType=IntegerType()), True)
])

Dalam contoh di atas kita mendefinisikan sebuah array dengan elemen bertipe integer.

Sedangkan contoh penggunaannya dalam DataFrame seperti di bawah ini

data = [['Agus Supono','F',[100,150,150]],
        ['Budi Sumardi','B',[200,100,150]],
        ['Dina Mariana','F',[150,150,130]],
        ['Dedi Setiadi','B', [50,100,100]]]


df = spark.createDataFrame(data, mySchema)
df.show()
df.printSchema()

+------------+-------+---------------+
|        nama|jurusan|          nilai|
+------------+-------+---------------+
| Agus Supono|      F|[100, 150, 150]|
|Budi Sumardi|      B|[200, 100, 150]|
|Dina Mariana|      F|[150, 150, 130]|
|Dedi Setiadi|      B| [50, 100, 100]|
+------------+-------+---------------+

root
 |-- nama: string (nullable = true)
 |-- jurusan: string (nullable = true)
 |-- nilai: array (nullable = true)
 |    |-- element: integer (containsNull = true)

Mengakses Kolom PySpark ArrayType

Untuk mengakses kolom bertipe array, kita gunakan perintah select seperti biasa

df.select("nama", "nilai").show()

+------------+---------------+
|        nama|          nilai|
+------------+---------------+
| Agus Supono|[100, 150, 150]|
|Budi Sumardi|[200, 100, 150]|
|Dina Mariana|[150, 150, 130]|
|Dedi Setiadi| [50, 100, 100]|
+------------+---------------+

Untuk mengakses salah satu elemen array menggunakan indeks, kita gunakan perintah pyspark.sql.functions.col(namakolom)[indeks]

Misalnya jika kita ingin mengakses kolom nama dan nilai pertama, alias array nilai di indeks ke 0, maka perintahnya adalah :

df.select("nama", col("nilai")[0]).show()

+------------+--------+
|        nama|nilai[0]|
+------------+--------+
| Agus Supono|     100|
|Budi Sumardi|     200|
|Dina Mariana|     150|
|Dedi Setiadi|      50|
+------------+--------+

Fungsi-fungsi PySpark ArrayType

Terdapat beberapa fungsi PySaprk yang terkait dengan ArrayType, yaitu array_contains(), explode(), split(), dan array()

Fungsi array_contains()

Fungsi ini digunakan untuk memeriksa apakah suatu nilai tertentu terdapat dalam array. Fungsi ini menerima argumen berupa kolom yang bertipe array, dan mengembalikan nilai berupa kolom bertipe boolean.

Misalnya pada contoh di atas, kita dapat memeriksa apakah kolom "nilai" yang bertipe array mengandung nilai 200. Hasilnya akan kita simpan ke dalam kolom bernama "mengandung_nilai_200"

df.select(df.nama,
          array_contains(df.nilai,200)
          .alias("mengandung_nilai_200")).show()

+------------+--------------------+
|        nama|mengandung_nilai_200|
+------------+--------------------+
| Agus Supono|               false|
|Budi Sumardi|                true|
|Dina Mariana|               false|
|Dedi Setiadi|               false|
+------------+--------------------+

Fungsi explode()

Fungsi explode digunakan untuk mengubah setiap elemen array menjadi satu baris atau record tersendiri. Misalnya untuk contoh di atas :

df.select("nama", "jurusan", explode(df["nilai"])).show()

+------------+-------+---+
|        nama|jurusan|col|
+------------+-------+---+
| Agus Supono|      F|100|
| Agus Supono|      F|150|
| Agus Supono|      F|150|
|Budi Sumardi|      B|200|
|Budi Sumardi|      B|100|
|Budi Sumardi|      B|150|
|Dina Mariana|      F|150|
|Dina Mariana|      F|150|
|Dina Mariana|      F|130|
|Dedi Setiadi|      B| 50|
|Dedi Setiadi|      B|100|
|Dedi Setiadi|      B|100|
+------------+-------+---+

Dari contoh di atas terlihat bahwa dari satu record, dihasilkan 3 record berbeda untuk masing-masing elemen nilai, dengan kolom nama dan jurusan yang sama.

Fungsi split()

Fungsi ini digunakan untuk membentuk array dari sebuah string. String dipecah berdasarkan karakter pemisah/delimiter tertentu, dan bagian-bagiannya dibentuk menjadi elemen dari sebuah array.

Misalnya dari contoh di atas, kita akan membuat kolom arrayNama dari kolom nama. Kita gunakan spasi sebagai delimiternya.

df.select("nama", "jurusan", "nilai",
          split(df.nama," ").alias("arrayNama")).show()

+------------+-------+---------------+---------------+
|        nama|jurusan|          nilai|      arrayNama|
+------------+-------+---------------+---------------+
| Agus Supono|      F|[100, 150, 150]| [Agus, Supono]|
|Budi Sumardi|      B|[200, 100, 150]|[Budi, Sumardi]|
|Dina Mariana|      F|[150, 150, 130]|[Dina, Mariana]|
|Dedi Setiadi|      B| [50, 100, 100]|[Dedi, Setiadi]|
+------------+-------+---------------+---------------+

Fungsi array()

Fungsi ini digunakan untuk membentuk kolom bertipe array dari gabungan kolom-kolom lain. Misalnya kita memiliki dataframe seperti berikut ini

data1 = [['Agus Supono','F',100,150,150],
        ['Budi Sumardi','B',200,100,150],
        ['Dina Mariana','F',150,150,130],
        ['Dedi Setiadi','B', 50,100,100]]


mycolumns = ["nama","jurusan","nilai1","nilai2","nilai3"]
df1 = spark.createDataFrame(data1, mycolumns)
df1.show()
df1.printSchema()

+------------+-------+------+------+------+
|        nama|jurusan|nilai1|nilai2|nilai3|
+------------+-------+------+------+------+
| Agus Supono|      F|   100|   150|   150|
|Budi Sumardi|      B|   200|   100|   150|
|Dina Mariana|      F|   150|   150|   130|
|Dedi Setiadi|      B|    50|   100|   100|
+------------+-------+------+------+------+

root
 |-- nama: string (nullable = true)
 |-- jurusan: string (nullable = true)
 |-- nilai1: long (nullable = true)
 |-- nilai2: long (nullable = true)
 |-- nilai3: long (nullable = true)

Kita dapat membentuk kolom nilai yang dibentuk dari kolom-kolom nilai1, nilai2, dan nilai3

df1.select(df1.nama, df1.jurusan,
           array(df1.nilai1,df1.nilai2,df1.nilai2)
           .alias("nilai")).show()

+------------+-------+---------------+
|        nama|jurusan|          nilai|
+------------+-------+---------------+
| Agus Supono|      F|[100, 150, 150]|
|Budi Sumardi|      B|[200, 100, 100]|
|Dina Mariana|      F|[150, 150, 150]|
|Dedi Setiadi|      B| [50, 100, 100]|
+------------+-------+---------------+

Wrapping Up

ArrayType adalah data kompleks yang digunakan untuk menyimpan data dalam bentuk array atau list. Fungsi-fungsi yang terkait dengan PySpark ArrayType,

array_contains()
explode(),
split()
array()

Notebook untuk tutorial ini dapat diakses di sini

Artikel seri Belajar PySpark sebelumnya