Daft: DataFrame Multimodal Terdistribusi buat ETL dan Analitik Skala Besar

Halo temen-temen, ketemu lagi sama aku, Ruby Abdullah. Kali ini aku mau ngajak kalian kenalan sama satu tool yang menurutku bakal makin sering kalian butuhin kalau kerjaan kalian mulai nyentuh data yang gede dan bentuknya macem-macem. Namanya Daft. Kalau kalian selama ini udah nyaman pakai pandas atau Polars buat ngolah tabel angka dan teks, Daft ini kayak versi yang jauh lebih ambisius: dia dirancang biar bisa ngolah bukan cuma angka dan teks, tapi juga gambar, embedding, URL, bahkan hasil inferensi model machine learning, semuanya di dalam satu DataFrame yang sama.

Yang bikin aku tertarik banget adalah kombinasi desainnya. Di permukaan, kamu nulis kode Python yang enak dibaca, mirip pandas. Tapi di dalemnya, mesin eksekusinya ditulis pakai Rust, jadi cepet dan hemat memori. Dan yang lebih keren lagi, Daft ini sifatnya lazy alias malas: dia nggak langsung ngitung apa-apa pas kamu nulis operasi, tapi nunggu sampai kamu bener-bener minta hasilnya. Berkat sifat lazy ini, Daft bisa nge-optimasi seluruh rangkaian operasi kamu dulu sebelum jalan, dan bahkan bisa nyebar kerjaan ke banyak mesin lewat Ray kalau datanya kegedean buat satu komputer.

Di tutorial ini aku bakal nemenin kalian dari nol. Kita mulai dari instalasi, terus bikin DataFrame pertama, ngerti konsep lazy execution dan .collect(), baca data dari Parquet, CSV, sampai S3, main-main sama expression dan UDF, ngolah kolom gambar dan URL (download, decode, resize), jalanin inferensi model di atas kolom, dan terakhir nge-scale-out pakai Ray. Santai aja, aku bakal jelasin pelan-pelan sambil kasih contoh kode yang bisa langsung kalian jalanin. Yuk kita mulai.

Introduction

Sebelum masuk ke kode, aku pengen kalian paham dulu kenapa Daft ini ada dan masalah apa yang dia selesein. Coba deh temen-temen bayangin skenario yang makin umum sekarang: kamu punya jutaan baris data yang tiap barisnya nggak cuma berisi angka, tapi juga path ke gambar, URL ke dokumen, potongan teks panjang, dan mungkin vektor embedding hasil dari model. Kalau kamu coba olah ini pakai pandas, kamu bakal cepet nabrak dua tembok. Tembok pertama adalah memori, karena pandas naruh semua data di RAM satu mesin. Tembok kedua adalah tipe data multimodal, karena pandas nggak punya konsep native buat kolom gambar atau tensor, jadi kamu ujung-ujungnya nyimpen object Python yang lambat dan berantakan.

Daft hadir buat ngejawab dua masalah ini sekaligus. Dari sisi skala, Daft bisa jalan di satu laptop buat data kecil, tapi juga bisa langsung disebar ke cluster Ray buat data yang ukurannya terabyte tanpa kamu harus nulis ulang kodenya. Dari sisi multimodal, Daft punya tipe data khusus buat gambar, tensor, embedding, dan URL, plus operasi bawaan buat download, decode, dan resize gambar. Jadi pipeline yang biasanya butuh gabungan pandas, PIL, requests, dan kode paralel manual, sekarang bisa ditulis rapi dalam satu API DataFrame.

Ada tiga konsep inti yang bakal terus muncul sepanjang tutorial ini, jadi mending aku kenalin dari awal. Pertama, DataFrame, yaitu struktur tabel utama tempat data kamu hidup. Kedua, Expression, yaitu cara kamu ngedeskripsiin komputasi di atas kolom tanpa langsung ngejalaninnya. Ketiga, lazy execution, yaitu prinsip bahwa Daft nunda semua komputasi sampai kamu panggil aksi kayak .collect() atau .show(). Tiga konsep ini yang bikin Daft beda dari pandas dan yang bikin dia bisa cepet dan skalabel.

Instalasi

Oke, sekarang kita mulai dari yang paling dasar dulu ya, yaitu masang Daft di mesin kalian. Satu hal yang kadang bikin orang bingung: nama package-nya di PyPI itu getdaft, bukan daft. Tapi pas kamu import di Python, kamu tetap nulis import daft. Jadi jangan kaget kalau perintah instalasinya keliatan beda dari nama import-nya.

Buat masang versi dasarnya, cukup jalanin perintah ini di terminal kalian.

pip install getdaft

Kalau kamu udah tahu bakal kerja sama data di cloud atau bakal nge-scale pakai Ray, aku saranin langsung pasang sekalian sama extras-nya biar nggak bolak-balik. Daft nyediain beberapa extras yang bisa kamu pilih sesuai kebutuhan.

# Support buat AWS S3, Google Cloud Storage, Azure
pip install "getdaft[aws]"

Support buat scaling pakai Ray
pip install "getdaft[ray]"

Pasang semua sekaligus
pip install "getdaft[all]"

Aku pribadi biasanya langsung pasang getdaft[all] kalau lagi di environment development, biar nggak ketemu error "module not found" di tengah jalan. Tapi kalau kamu lagi bangun image Docker yang pengen ramping, mending pilih extras yang bener-bener kamu butuh aja.

Setelah kepasang, ada baiknya kita tes dulu bahwa semuanya jalan. Coba jalanin skrip kecil ini.

import daft

print(daft.version)

df = daft.frompydict({

    "nama": ["Ruby", "Sinta", "Bagas"],
    "umur": [28, 24, 31],
})
df.show()

Kalau kamu ngeliat versi Daft ke-print dan sebuah tabel kecil muncul di layar, berarti instalasi kamu udah beres dan kita siap lanjut. Fungsi daft.frompydict di atas itu cara paling gampang bikin DataFrame dari dictionary Python, di mana key jadi nama kolom dan value jadi isi kolomnya.

Basic Usage

Bikin DataFrame Pertama

Ada banyak cara bikin DataFrame di Daft, dan yang paling sering aku pakai buat eksperimen cepet adalah frompydict. Tapi selain itu kamu juga bisa bikin dari list of dictionaries pakai frompylist, atau langsung dari pandas DataFrame yang udah kamu punya. Coba lihat beberapa cara ini.

import daft

Dari dictionary of columns
df1 = daft.frompydict({

    "id": [1, 2, 3],
    "produk": ["Kopi", "Teh", "Susu"],
    "harga": [25000, 18000, 22000],
})

Dari list of rows
df2 = daft.frompylist([
    {"id": 1, "produk": "Kopi", "harga": 25000},
    {"id": 2, "produk": "Teh", "harga": 18000},
])

Dari pandas
import pandas as pd
pdf = pd.DataFrame({"id": [1, 2], "produk": ["Kopi", "Teh"]})
df3 = daft.frompandas(pdf)


df1.show()

Nah, ini bagian yang paling penting buat kalian pahami, jadi tolong baca pelan-pelan ya temen-temen. Di Daft, pas kamu nulis operasi kayak select, filter, atau withcolumn, Daft nggak langsung ngejalanin komputasi itu. Yang dia lakuin cuma nyatet rencana atau plan tentang apa yang mau kamu kerjain. Komputasi baru bener-bener jalan pas kamu manggil sebuah aksi, dan aksi yang paling umum adalah .collect().

Kenapa desainnya begini? Karena dengan nunggu, Daft bisa ngeliat seluruh rangkaian operasi kamu sekaligus dan nge-optimasinya. Misalnya kalau kamu baca file gede tapi cuma butuh dua kolom dan cuma baris yang harganya di atas 20000, Daft cukup pinter buat cuma baca kolom yang perlu dan nyaring baris seawal mungkin, jadi kerjaannya jauh lebih ringan.

Coba lihat contoh ini biar kebayang bedanya.

import daft

df = daft.frompydict({

    "produk": ["Kopi", "Teh", "Susu", "Jus"],
    "harga": [25000, 18000, 22000, 30000],
})

Ini semua masih lazy, belum ada komputasi yang jalan
mahal = df.where(df["harga"] > 20000).select("produk", "harga")

Baru di sini komputasi jalan dan hasilnya di-materialize
hasil = mahal.collect()
print(hasil)

Di kode di atas, baris mahal = ... itu cuma nyusun rencana. Baru pas mahal.collect() dipanggil, Daft jalanin semuanya dan naruh hasilnya ke memori. Selain .collect(), ada beberapa aksi lain yang juga men-trigger eksekusi, kayak .show(n) buat nampilin beberapa baris pertama, .topandas() buat ngubah ke pandas, .topydict() buat ngubah ke dictionary, dan .writeparquet() buat nulis ke disk. Selama kamu belum manggil salah satu dari aksi ini, DataFrame kamu masih berupa rencana yang belum dieksekusi.

Satu tips dari aku: kalau kamu lagi ngembangin pipeline dan pengen ngintip hasil sementara, pakai .show(5) aja, jangan .collect() semua, apalagi kalau datanya gede. .show() cuma ngambil beberapa baris jadi jauh lebih cepet dan hemat memori.

Filter, Select, dan Kolom Baru

Operasi dasar buat ngolah tabel di Daft mirip banget sama library DataFrame lain, tapi dengan gaya expression yang khas. Buat milih kolom pakai select, buat nyaring baris pakai where, dan buat bikin kolom baru pakai withcolumn. Yuk kita gabungin semuanya.

import daft

df = daft.frompydict({
    "produk": ["Kopi", "Teh", "Susu", "Jus"],
    "harga": [25000, 18000, 22000, 30000],
    "stok": [10, 25, 8, 15],
})

hasil = (
    df
    .withcolumn("hargadiskon", df["harga"]  0.9)

    .withcolumn("nilaistok", df["harga"]  df["stok"])
    .where(df["stok"] > 9)
    .select("produk", "hargadiskon", "nilaistok")
)

hasil.show()

Perhatiin gaya nulisnya yang berantai atau chaining. Tiap operasi ngembaliin DataFrame baru, jadi kamu bisa terus nyambungin operasi berikutnya. Ekspresi kayak df["harga"] 0.9 itu contoh Expression, yaitu deskripsi komputasi di atas kolom yang belum dieksekusi. Aku bakal bahas expression lebih dalam di bagian advanced.

import daft

df = daft.frompydict({

    "kategori": ["minuman", "makanan", "minuman", "makanan", "minuman"],
    "harga": [25000, 40000, 18000, 35000, 30000],
})

ringkasan = df.groupby("kategori").agg(
    df["harga"].mean().alias("rataharga"),
    df["harga"].sum().alias("totalharga"),

    df["harga"].count().alias("jumlahitem"),
)

ringkasan.show()

import daft

Baca satu file atau banyak file sekaligus pakai wildcard
dfcsv = daft.readcsv("data/penjualan.csv")
dfparquet = daft.readparquet("data/.parquet")

Baca langsung dari S3
dfs3 = daft.readparquet("s3://bucket-ku/dataset/.parquet")


dfparquet.show()

import daft
from daft.io import IOConfig, S3Config

ioconfig = IOConfig(
    s3=S3Config(
        keyid="AKIA...",

        accesskey="rahasia...",
        regionname="ap-southeast-1",

    )
)

df = daft.readparquet("s3://bucket-ku/data/.parquet", ioconfig=ioconfig)
df.show()

Ingat, karena Daft itu lazy, readparquet di atas nggak langsung nyedot seluruh file ke memori. Daft cuma baca metadata dan skema-nya dulu. Data baru dibaca beneran pas kamu panggil aksi, dan cuma kolom serta baris yang perlu aja yang dibaca berkat optimasi yang tadi aku ceritain.

Expression adalah jantungnya Daft. Tiap kali kamu nulis komputasi di atas kolom, kamu sebenernya lagi nyusun expression. Cara paling umum ngeakses kolom adalah df["namakolom"], tapi kamu juga bisa pakai daft.col("namakolom") yang berguna kalau kamu belum punya referensi DataFrame-nya. Expression punya banyak method yang dikelompokin dalam namespace, misalnya .str buat operasi string, .dt buat tanggal, .image buat gambar, dan .url buat URL.

import daft

df = daft.frompydict({
    "nama": ["ruby abdullah", "sinta dewi", "bagas pratama"],
    "email": ["ruby@mail.com", "sinta@mail.com", "bagas@mail.com"],
})

hasil = df.select(
    daft.col("nama").str.upper().alias("namakapital"),

    daft.col("nama").str.length().alias("panjangnama"),
    daft.col("email").str.split("@").alias("bagianemail"),

)

hasil.show()

import daft

@daft.udf(returndtype=daft.DataType.string())
def kategoriharga(harga):

    hasil = []
    for h in harga.topylist():
        if h < 20000:
            hasil.append("murah")
        elif h < 28000:
            hasil.append("sedang")
        else:
            hasil.append("mahal")
    return hasil

df = daft.frompydict({

    "produk": ["Kopi", "Teh", "Jus"],
    "harga": [25000, 18000, 30000],
})

df = df.withcolumn("segmen", kategoriharga(df["harga"]))

df.show()

Perhatiin dua hal penting. Pertama, kamu wajib nyebut returndtype biar Daft tahu tipe data kolom hasilnya. Kedua, argumen harga yang masuk ke fungsi itu bukan satu angka, tapi sebuah series, makanya aku panggil .topylist() buat iterasi. Pola ini emang butuh sedikit adaptasi kalau kamu terbiasa sama apply di pandas yang per baris, tapi imbalannya performa yang jauh lebih baik.

import daft

df = daft.frompydict({
    "url": [
        "https://example.com/foto1.jpg",
        "https://example.com/foto2.jpg",
        "https://example.com/foto3.jpg",
    ],
})

df = (
    df
    .withcolumn("bytes", df["url"].url.download(onerror="null"))
    .withcolumn("gambar", daft.col("bytes").image.decode())

    .withcolumn("kecil", daft.col("gambar").image.resize(224, 224))
)

df.select("url", "kecil").show()

Beberapa hal yang perlu kalian catat. Argumen onerror="null" di url.download itu penyelamat, karena pas kamu download ribuan URL pasti ada yang gagal atau mati, dan dengan setting ini baris yang gagal bakal jadi null bukannya bikin seluruh pipeline crash. Terus, image.resize(224, 224) itu contoh nyeragamin ke ukuran 224x224, yang kebetulan ukuran input umum buat banyak model vision. Semua operasi ini tetep lazy dan bakal jalan paralel pas kamu collect, jadi kamu dapet download dan decode konkuren gratis tanpa nulis kode threading sendiri.

import daft
import numpy as np

@daft.udf(returndtype=daft.DataType.python())
class KlasifikasiGambar:
    def init(self):
        # Load model sekali aja pas worker dinisialisasi
        # Contoh pseudo: ganti dengan model beneran kamu
        self.model = loadmodelkamu()

    def call(self, kolomgambar):

        gambarbatch = kolomgambar.topylist()
        prediksi = []
        for img in gambarbatch:

            arr = np.array(img)
            hasil = self.model.predict(arr[None, ...])
            prediksi.append(hasil.tolist())
        return prediksi

df = daft.frompydict({"url": ["https://example.com/a.jpg"]})
df = (
    df
    .withcolumn("gambar", df["url"].url.download(onerror="null").image.decode())
    .withcolumn("prediksi", KlasifikasiGambar(daft.col("gambar")))

)
df.show()

Buat model yang butuh GPU, Daft juga ngasih cara nge-request resource. Kamu bisa nambahin argumen kayak numgpus=1 pas mendeklarasi UDF, biar scheduler tahu bahwa UDF ini butuh GPU dan ngatur penempatannya di cluster dengan bener.

@daft.udf(returndtype=daft.DataType.python(), numgpus=1)
class ModelGPU:
    def init(self):
        self.model = loadmodelkegpu()


    def call(self, kolom):
        return [self.model(x) for x in kolom.topylist()]

Scaling Out dengan Ray

Sampai sini semua kode yang kita tulis jalan di satu mesin pakai runner lokal Daft. Tapi kekuatan sejati Daft muncul pas datanya udah kegedean buat satu mesin. Di titik ini, kamu bisa nyuruh Daft nyebar seluruh kerjaan ke cluster Ray, dan yang bikin aku takjub, kamu nggak perlu ngubah kode pipeline kamu sama sekali. Cukup ganti runner-nya di awal skrip.

import daft

Sambungin Daft ke cluster Ray
daft.context.setrunnerray(address="ray://kepala-cluster:10001")

Kalau Ray jalan lokal, cukup tanpa address
daft.context.setrunnerray()

df = daft.readparquet("s3://bucket-ku/dataset-gede/*.parquet")

df = df.where(df["harga"] > 20000).groupby("kategori").agg(
    df["harga"].mean().alias("rataharga")
)
df.collect()

Perhatiin bahwa satu-satunya baris baru adalah daft.context.setrunnerray(). Seluruh operasi DataFrame di bawahnya persis sama kayak yang kamu tulis buat mesin lokal. Inilah janji utama Daft: kamu nulis logika sekali, dan dia yang ngurusin apakah jalannya di satu laptop atau di seratus mesin. Berkat sifat lazy tadi, Daft bisa mecah kerjaan kamu jadi banyak task kecil dan nyebarnya ke seluruh worker Ray secara otomatis.

Ini juga alasan kenapa pola-pola yang tadi kita bahas, kayak download URL paralel dan class UDF yang load model sekali per worker, jadi makin bermakna di skala cluster. Tiap worker Ray narik bagian datanya, download gambarnya, jalanin modelnya, dan ngirim hasil balik, semuanya konkuren tanpa kamu nulis satu baris kode distribusi pun.

Best Practices

Setelah aku pakai Daft buat beberapa proyek, ada beberapa kebiasaan yang menurutku bikin hidup lebih gampang. Aku bagi ke kalian biar kalian nggak perlu ngalamin trial and error yang sama.

Pertama, manfaatin sifat lazy sepenuhnya. Jangan buru-buru panggil .collect() di tengah pipeline cuma buat ngintip. Susun seluruh rangkaian transformasi kamu dulu sampai selesai, baru collect di akhir. Kalau cuma pengen ngecek bentuk data, pakai .show(5) yang cuma narik beberapa baris. Dengan begini kamu ngasih Daft kesempatan maksimal buat nge-optimasi seluruh rencana sekaligus.

Kedua, selalu tentuin returndtype yang tepat di UDF kamu. Ini bukan cuma formalitas, tapi ngebantu Daft ngatur memori dan validasi tipe dengan bener. Kalau outputnya struktur kompleks yang nggak punya tipe Daft native, baru pakai daft.DataType.python(), tapi kalau bisa pakai tipe konkret kayak string, integer, atau list, mendingan pakai itu karena lebih efisien.

Ketiga, buat operasi download dan I/O yang rawan gagal, selalu set onerror="null". Pas kamu ngolah ribuan atau jutaan URL, dijamin ada yang mati, timeout, atau formatnya rusak. Kamu nggak mau satu URL rusak ngejatuhin pipeline yang udah jalan berjam-jam. Dengan onerror="null", baris bermasalah jadi null dan kamu bisa nyaringnya belakangan pakai where.

Kelima, kembangin di lokal, deploy di cluster. Waktu ngembangin, pakai subset data kecil dan runner lokal biar iterasi kamu cepet. Setelah pipeline-nya bener, baru ganti ke setrunnerray() dan arahin ke data penuh. Karena kodenya identik, transisi ini mulus dan kamu nggak perlu debug ulang logika bisnis di cluster yang mahal.

Keenam, perhatiin ukuran batch dan resource pas kerja sama gambar dan model. Gambar itu makan memori banyak, apalagi kalau resolusinya gede. Resize seawal mungkin dalam pipeline biar data yang ngalir ke tahap berikutnya lebih ringan. Dan buat model GPU, deklarasiin numgpus dengan bener biar scheduler nempatin task di mesin yang punya GPU.

Ketujuh, pahami kapan harus pakai .topandas(). Daft gampang banget diubah ke pandas buat langkah akhir kayak plotting atau integrasi sama library yang belum support Daft. Tapi ingat, .topandas() narik semua data ke satu mesin, jadi jangan ngelakuin ini di data yang gede. Lakuin agregasi atau filtering berat di Daft dulu sampai datanya ciut, baru convert ke pandas di ujung.

Conclusion

Oke temen-temen, kita udah nempuh perjalanan yang lumayan panjang dari nol sampai bisa nge-scale pipeline ke cluster. Aku harap sekarang kalian punya gambaran yang jelas tentang kenapa Daft ini spesial dan kapan kalian bakal butuh dia.

Poin utama yang mau aku tekanin: Daft itu jawaban buat dua masalah yang makin sering muncul di era data modern, yaitu skala yang besar dan data yang multimodal. Berkat API Python yang ramah dan mesin Rust yang cepet, plus desain lazy yang bikin dia bisa nge-optimasi dan nyebar kerjaan, kamu bisa nulis pipeline yang enak dibaca tapi tetap sanggup ngolah terabyte data berisi gambar, teks, embedding, dan URL. Dan yang paling aku suka, kamu nulis logikanya sekali, terus tinggal ganti runner buat lompat dari laptop ke cluster Ray tanpa nulis ulang apa-apa.

Saran aku, mulai dari yang kecil dulu. Coba pasang getdaft, bikin DataFrame dari dictionary, mainin filter dan expression, terus pelan-pelan naik ke download gambar dan UDF. Setelah kamu nyaman, baru sentuh Ray buat scaling. Belajar bertahap gini bikin konsep lazy execution dan expression benerbener nempel, dan itu fondasi yang bakal ngebantu kamu di semua fitur lanjutan Daft.

Segitu dulu dari aku. Kalau kalian ngerasa tutorial ini ngebantu, cobain langsung di proyek kalian dan rasain sendiri bedanya ngolah data multimodal pakai tool yang emang dirancang buat itu. Sampai ketemu di tutorial berikutnya, dan selamat ngoprek temen-temen.