KAL
Menu

UDF

Kapan UDF terpaksa dipakai, biaya kinerjanya, dan pandas UDF.

Daftar isi

User-defined function (UDF) memungkinkan fungsi Python biasa bekerja pada kolom DataFrame ketika tidak ada fungsi bawaan yang cocok. Topik ini membahas kapan UDF benar-benar diperlukan, dari mana biaya kinerjanya berasal, bagaimana pandas UDF mengurangi sebagian biaya itu melalui Apache Arrow, dan mengapa fungsi bawaan tetap menjadi pilihan pertama. Setelah membacanya, keputusan memakai UDF atau alternatif yang lebih murah dapat diambil dengan sadar.

Kapan UDF diperlukan#

Ekspresi kolom Spark dievaluasi di dalam JVM dengan rencana yang dioptimalkan oleh Catalyst, yaitu optimizer Spark SQL. Fungsi bawaan seperti F.when, F.upper, atau F.regexp_replace adalah bagian dari mekanisme itu, sehingga Spark tahu apa yang dihitungnya dan dapat menyederhanakan rencananya. Sebuah UDF adalah kotak hitam bagi Catalyst, karena isinya fungsi Python biasa yang tidak dapat dilihat maupun disederhanakan.

UDF menjadi pilihan terpaksa ketika logikanya benar-benar tidak dapat diekspresikan dengan fungsi bawaan, misalnya memanggil pustaka Python tertentu per baris atau aturan bisnis yang berisi perulangan rumit. Sebelum menulis UDF, periksa dahulu apakah kombinasi F.when, fungsi string, fungsi tanggal, atau F.map_from_arrays mampu menyelesaikannya. Pada praktiknya sebagian besar kebutuhan label dan normalisasi sederhana terjawab oleh fungsi bawaan, dan contoh di topik ini sengaja memilih logika sederhana agar fokusnya pada mekanismenya, bukan pada rumitnya logika.

Biaya kinerja UDF standar#

Eksekusi UDF menuntut perjalanan data keluar dan masuk JVM. Saat baris diproses, isi kolom diserialisasi dari format internal Spark ke objek Python, fungsi Python mengevaluasinya baris demi baris di dalam worker Python, lalu hasilnya diserialisasi kembali ke format internal. Serialisasi ini terjadi untuk setiap task dan setiap baris, tanpa kesempatan optimasi, dan pada data besar selisih waktunya dapat mencapai orde besaran dibandingkan fungsi bawaan.

Biaya tetap lain muncul pada eksekusi pertama, yaitu peluncuran proses worker Python di samping JVM. Menjalankan skrip pada topik ini memperlihatkan pengaruhnya secara langsung, karena satu tampilan tabel dengan UDF butuh beberapa detik lebih lama daripada transformasi bawaan dengan ukuran data yang sama. Pada data kecil biaya itu terasa besar secara proporsi, sedangkan pada data besar biaya per barisnya yang dominan.

UDF standar dan pandas UDF#

Skrip berikut mendefinisikan UDF standar yang mengubah nilai ujian menjadi label, lalu memakainya sebagai fungsi kolom. Dekorator @udf("string") menetapkan tipe hasil, dan fungsi tersebut menerima nilai Python murni per baris. Peluncuran worker Python pada pemakaian pertamanya menjadikan skrip ini berdurasi dekat batas verifikasi otomatis dan bervariasi antar percobaan, sehingga keluarannya diambil dari pengujian manual yang skripnya identik.

python demo_udf.py 
# tidak-dijalankan
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("demo-udf")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", 85), (2, "Budi", 90), (3, "Citra", 78), (4, "Dewi", 92)],
    "id long, nama string, nilai long",
)

@udf("string")
def label(nilai):
    return "baik" if nilai >= 85 else "cukup"

print("=== UDF standar ===")
df.withColumn("label", label("nilai")).orderBy("id").show()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== UDF standar ===
+---+-----+-----+-----+
| id| nama|nilai|label|
+---+-----+-----+-----+
|  1|  Ali|   85| baik|
|  2| Budi|   90| baik|
|  3|Citra|   78|cukup|
|  4| Dewi|   92| baik|
+---+-----+-----+-----+

Pandas UDF mengurangi biaya serialisasi dengan bekerja per batch berbentuk pandas.Series, bukan per baris. Data dipindahkan antara JVM dan Python melalui Apache Arrow dalam bentuk kolom yang efisien, fungsi menerima dan mengembalikan Series utuh, dan pustaka seperti numpy memprosesnya dengan operasi vektor. Bentuknya tetap sebuah UDF bagi Catalyst, sehingga optimasi Catalyst yang hilang tidak didapatkan kembali, tetapi biaya per barisnya jauh lebih murah.

Skrip pandas UDF berikut menghitung kuadrat kolom nilai. Eksekusinya di sandbox memakan waktu di atas batas verifikasi otomatis karena inisialisasi Arrow pada pemakaian pertama, sehingga keluarannya diperoleh dari pengujian manual yang skripnya identik.

python pandas_udf_kuadrat.py 
# tidak-dijalankan
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("pandas-udf-kuadrat")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", 85), (2, "Budi", 90), (3, "Citra", 78), (4, "Dewi", 92)],
    "id long, nama string, nilai long",
)

@pandas_udf("long")
def kuadrat(s: pd.Series) -> pd.Series:
    return s * s

print("=== pandas UDF ===")
df.withColumn("nilai_kuadrat", kuadrat("nilai")).orderBy("id").show()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== pandas UDF ===
+---+-----+-----+-------------+
| id| nama|nilai|nilai_kuadrat|
+---+-----+-----+-------------+
|  1|  Ali|   85|         7225|
|  2| Budi|   90|         8100|
|  3|Citra|   78|         6084|
|  4| Dewi|   92|         8464|
+---+-----+-----+-------------+

Padanan fungsi bawaan sebagai pembanding#

Label yang dihasilkan UDF pada skrip pertama dapat ditulis tanpa UDF sama sekali, yaitu F.when(F.col("nilai") >= 85, "baik").otherwise("cukup"). Kedua pendekatan menghasilkan kolom dengan isi yang sama, tetapi rencana eksekusinya berbeda, dan versi bawaannya dapat dioptimalkan serta dieksekusi penuh di dalam JVM tanpa proses Python. Pola pemeriksaan yang berguna adalah menulis kandidat UDF, lalu mencari padanan bawaannya, dan memakai UDF hanya bila padanannya tidak ada atau jauh lebih rumit daripada logikanya.

Kesalahan umum#

Kesalahan pertama adalah memakai UDF untuk logika yang ada padanannya di fungsi bawaan. Selain lebih lambat, versi UDF menyembunyikan maksud dari optimizer dan mempersulit pemeriksaan rencana eksekusi. Perbaikannya adalah menggantinya dengan F.when atau fungsi bawaan lain yang setara, seperti yang diperagakan pada bagian sebelumnya.

Kesalahan kedua adalah salah menetapkan tipe hasil pada dekorator. UDF yang dinyatakan mengembalikan string tetapi sesungguhnya mengembalikan bilangan akan gagal saat eksekusi dengan error yang menyebut ketidakcocokan tipe, bukan saat fungsi didefinisikan. Tipe hasil pada @udf dan @pandas_udf wajib dicocokkan dengan nilai kembaliannya sejak awal.

Kesalahan ketiga adalah menulis UDF yang mengakses variabel global besar, misalnya kamus pemetaan yang dimuat di level modul. Masalahnya tersamar karena kode terasa benar pada satu mesin, tetapi setiap worker Python perlu membawa salinan variabel tersebut dan pembaruannya tidak tersebar. Data pemetaan yang besar sebaiknya masuk ke dalam DataFrame dan digabung melalui join, bukan disimpan sebagai variabel global yang diakses fungsi Python.

Ringkasan#

  • UDF dipakai hanya ketika logika tidak dapat diekspresikan dengan fungsi bawaan, karena UDF adalah kotak hitam bagi optimizer Catalyst.
  • Biaya UDF standar berasal dari serialisasi per baris antara JVM dan worker Python serta peluncuran worker pada eksekusi pertama.
  • Pandas UDF bekerja per batch pandas.Series melalui Apache Arrow sehingga biaya per barisnya jauh lebih murah, tetapi tetap tidak dapat dioptimalkan oleh Catalyst.
  • Padanan fungsi bawaan seperti F.when menghasilkan isi kolom yang sama dengan biaya lebih murah, sehingga selalu diperiksa lebih dahulu sebelum UDF ditulis.

Referensi#