Meningkatkan Alur Kerja Data Science Anda dengan Dask: Panduan Komprehensif

Saat Wes McKinney mulai menulis pandas, ia punya aturan praktis: agar pandas bekerja optimal, ukuran RAM mesin harus 5–10 kali lebih besar daripada dataset yang digunakan. Aturan ini mudah diikuti sekitar tahun 2010, tetapi sekarang sudah 2023.

Pada 2020, dataset dunia nyata sudah berkembang hingga ukuran yang dapat dengan mudah membuat laptop dan mesin sehari-hari crash. Mengantisipasi masalah ini jauh hari, sebuah solusi dirilis pada 2015.

Dask adalah pustaka open-source yang dikembangkan oleh pembuat Anaconda untuk mengatasi tantangan komputasi yang skalabel dan efisien pada dataset besar yang melebihi kapasitas memori satu mesin.

Tutorial ini memberikan pengenalan komprehensif tentang Dask dan fitur-fiturnya yang krusial, termasuk antarmuka untuk DataFrame, Array, dan Bag.

Menyiapkan Dask

Seperti pustaka lainnya, Anda dapat menginstal Dask dengan tiga cara: Conda, Pip, dan dari source.

Karena ini adalah artikel pengantar tentang Dask, kita tidak akan membahas metode instalasi terakhir karena ditujukan untuk maintainer.

Jika Anda menggunakan Anaconda, Dask sudah termasuk dalam instalasi default Anda (yang menandakan betapa populernya pustaka ini). Jika Anda ingin memasang ulang atau memutakhirkannya, Anda dapat menggunakan perintah install:

conda install dask

Alternatif melalui PIP sebagai berikut:

pip install "dask[complete]"

Menambahkan ekstensi [complete] juga memasang dependensi yang diperlukan Dask, sehingga tidak perlu memasang NumPy, Pandas, dan Tornado secara manual.

Anda dapat memeriksa apakah instalasi berhasil dengan melihat versi pustaka:

import dask

dask.__version__

Keluaran:

'2023.5.0'

Sebagian besar pekerjaan Anda dengan Dask akan berfokus pada tiga antarmuka: Dask DataFrame, Array, dan Bag. Mari kita impor semuanya bersama numpy dan pandas untuk digunakan sepanjang artikel:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pd

Konsep Dasar Dask

Secara tingkat tinggi, Anda dapat menganggap Dask sebagai pembungkus yang memperluas kapabilitas alat tradisional seperti pandas, NumPy, dan Spark untuk menangani dataset yang lebih besar daripada memori.

Saat berhadapan dengan objek besar seperti array (vektor) atau matriks (dataframe) yang lebih besar dari memori, Dask memecahnya menjadi potongan-potongan kecil, yang juga disebut partisi.

Sebagai contoh, perhatikan array berisi 12 angka acak di NumPy dan Dask:

narr = np.random.rand(12)

narr

array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])

darr = da.from_array(narr, chunks=3)
darr

Gambar di atas menunjukkan bahwa array Dask berisi empat chunk karena kita menetapkan chunks ke 3. Di balik layar, setiap chunk adalah array NumPy tersendiri.

Sekarang, mari pertimbangkan contoh yang jauh lebih besar. Kita akan membuat dua array 10k x 100k (1 miliar elemen) dan melakukan perkalian elemen demi elemen di kedua pustaka sambil mengukur kinerjanya:

# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))

%%time

result_np = np.multiply(arr1, arr2)

Wall time: 3.19 s

%%time

result_dask = da.multiply(darr1, darr2)

Wall time: 94.8 ms

Kode di atas menunjukkan perkalian elemen demi elemen dari dua array besar menggunakan NumPy dan Dask. Seperti terlihat pada keluaran, Dask kira-kira 34 kali lebih cepat daripada NumPy untuk komputasi ini. Peningkatan kinerja menjadi semakin signifikan seiring meningkatnya kompleksitas komputasi dan ukuran array.

Dask menggunakan pendekatan serupa, yakni memecah menjadi chunk dan mendistribusikannya ke semua inti yang tersedia pada mesin Anda untuk objek-objek lain juga.

Dask DataFrame

Untuk benar-benar menghargai manfaat Dask, kita memerlukan dataset besar, sebaiknya berukuran di atas 1 GB. Namun, mengunduh dataset seperti itu untuk diikuti dalam tutorial mungkin kurang optimal. Sebagai gantinya, Anda dapat menggunakan skrip ini, yang menghasilkan dataset sintetis dengan 10 juta baris, 10 fitur numerik, dan 10 fitur kategorikal.

Pastikan mesin Anda memiliki setidaknya 12 GB RAM untuk menjalankan skrip.

Setelah Anda memiliki file large_dataset.csv di workspace, Anda dapat memuatnya menggunakan fungsi read_csv dari antarmuka Dask DataFrame (dd):

import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()

Meskipun filenya besar, Anda akan melihat hasil diambil hampir seketika. Untuk file yang lebih besar lagi, Anda dapat menentukan parameter `blocksize`, yang menentukan jumlah byte untuk memecah file.

Mirip dengan bagaimana Dask Array berisi chunk dari array NumPy kecil, Dask dirancang untuk menangani banyak DataFrame Pandas kecil yang diatur sepanjang indeks baris.

Seperti yang bisa Anda duga dari fungsi read_csv, sebagian besar sintaks dan fungsionalitas API pandas yang umum digunakan dipertahankan di Dask. Blok kode berikut semestinya sudah familier bagi Anda dari pengalaman bekerja dengan pandas.

Memilih kolom dan operasi elemen demi elemen

Dalam contoh ini, kita melakukan operasi kolom yang cukup lugas pada Dask DataFrame bernama dask_df. Kita menjumlahkan nilai dari kolom Numeric_0 dengan hasil perkalian nilai dari Numeric_9 dan Numeric_3. Hasilnya kita simpan dalam variabel bernama result.

result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64

Seperti yang telah kami sebutkan, Dask sedikit berbeda dari alat komputasi tradisional karena tidak langsung mengeksekusi operasi ini. Alih-alih, Dask membuat semacam 'rencana' yang disebut grafik tugas (task graph) untuk menjalankan operasi tersebut nanti. Pendekatan ini memungkinkan Dask mengoptimalkan komputasi dan melakukan paralelisasi saat diperlukan. Fungsi compute() memicu Dask untuk akhirnya menjalankan komputasi ini, dan head() hanya menampilkan beberapa baris pertama dari hasilnya.

Penyaringan bersyarat

Sekarang, mari lihat bagaimana Dask dapat menyaring data. Kita memilih baris dari DataFrame tempat nilai pada kolom "Categorical_5" adalah "A".

Proses penyaringan ini mirip dengan di pandas, tetapi dengan perbedaan — Dask melakukannya secara lazy. Dask menyiapkan task graph untuk operasi ini tetapi menunggu eksekusi hingga kita memanggil compute(). Saat kita menjalankan head(), kita dapat melihat beberapa baris pertama dari DataFrame yang telah disaring.

dask_df[dask_df["Categorical_5"] == "A"].compute().head()

Statistik ringkasan umum

Berikutnya, kita akan membuat beberapa statistik ringkasan umum menggunakan fungsi describe() milik Dask.

Fungsi ini memberikan sejumlah statistik deskriptif untuk DataFrame kita, termasuk mean, simpangan baku, nilai minimum, maksimum, dan seterusnya. Seperti pada contoh sebelumnya, Dask menyiapkan task graph untuk operasi ini ketika kita memanggil describe(), tetapi menunggu eksekusi hingga kita memanggil compute().

dask_df.describe().compute()

dask_df["Categorical_3"].value_counts().compute().head()

Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64

Kita juga menggunakan value_counts() untuk menghitung jumlah kemunculan tiap nilai unik pada kolom "Categorical_3". Kita memicu operasinya dengan compute(), dan head() menampilkan nilai yang paling umum.

Groupby

Terakhir, mari gunakan fungsi groupby() untuk mengelompokkan data berdasarkan nilai pada kolom "Categorical_8". Lalu kita pilih kolom "Numeric_7" dan menghitung mean untuk tiap kelompok.

Ini mirip dengan penggunaan ‘groupby()’ di pandas, tetapi seperti yang mungkin sudah Anda duga, Dask melakukannya secara lazy. Kita memicu operasi dengan compute(), dan head() menampilkan rata-rata kolom "Numeric_7" untuk beberapa kelompok pertama.

dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64

Lihat juga bagian dari panduan pengguna Dask ini tentang kesamaan lain antara pandas dan Dask.

Lazy evaluation

Sekarang, mari kita telusuri penggunaan fungsi compute di akhir setiap blok kode.

Dask mengevaluasi blok kode dalam mode lazy dibandingkan mode eager milik Pandas, yang langsung mengembalikan hasil.

Sebagai analogi memasak, lazy evaluation seperti menyiapkan bahan dan memotong sayuran terlebih dahulu tetapi baru mencampur untuk dimasak saat dibutuhkan. Fungsi compute menjalankan peran tersebut.

Sebaliknya, eager evaluation seperti langsung memasukkan bahan ke wajan untuk dimasak begitu siap. Pendekatan ini memastikan semuanya siap disajikan sekaligus.

Lazy evaluation adalah kunci performa unggul Dask karena memberikan:

1. Pengurangan komputasi. Ekspresi hanya dievaluasi saat diperlukan (ketika compute dipanggil), menghindari hasil antara yang tidak diperlukan dan mungkin tidak digunakan dalam hasil akhir.
2. Alokasi sumber daya yang optimal. Lazy evaluation menghindari pengalokasian memori atau daya proses untuk hasil antara yang mungkin tidak dibutuhkan.
3. Dukungan untuk dataset besar. Metode ini memproses elemen data secara on-the-fly atau dalam chunk yang lebih kecil, memungkinkan pemanfaatan memori yang efisien.

Saat hasil compute dikembalikan, hasil diberikan sebagai Series/DataFrame Pandas atau array NumPy alih-alih DataFrame Dask native.

>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series

Alasannya, sebagian besar operasi manipulasi data hanya mengembalikan subset dari dataframe asli, yang memakan ruang jauh lebih kecil. Jadi, tidak perlu menggunakan paralelisme Dask, dan Anda dapat melanjutkan alur kerja berikutnya di pandas atau NumPy.

Dask Bag dan Dask Delayed untuk Data Tidak Terstruktur

Dask Bag dan Dask Delayed adalah dua komponen pustaka Dask yang menyediakan alat kuat untuk bekerja dengan data tidak terstruktur atau semi-terstruktur serta memungkinkan lazy evaluation.

Jika dahulu data tabular paling umum, dataset saat ini sering melibatkan berkas tidak terstruktur seperti gambar, berkas teks, video, dan audio. Dask Bag menyediakan fungsionalitas dan API untuk menangani berkas tidak terstruktur tersebut secara paralel dan skalabel.

Sebagai contoh, mari pertimbangkan ilustrasi sederhana:

import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']

Dalam contoh ini, kita membuat Dask Bag b dari daftar string. Lalu kita menerapkan operasi pada Bag untuk menyaring string yang diawali huruf 'a' dan mengubahnya menjadi huruf kapital menggunakan fungsi filter() dan map(). Terakhir, kita hitung hasilnya sebagai daftar dengan metode compute() dan mencetak keluarannya.

Sekarang bayangkan Anda dapat melakukan operasi yang lebih kompleks pada miliaran string serupa yang disimpan dalam berkas teks. Tanpa lazy evaluation dan paralelisme yang ditawarkan Dask Bag, Anda akan menghadapi tantangan besar. (Baca lebih lanjut tentang Bag di dokumentasi Dask).

Adapun Dask Delayed, ia memberikan fleksibilitas yang lebih besar dan menghadirkan lazy evaluation serta paralelisme ke berbagai skenario lainnya. Dengan Dask Delayed, Anda dapat mengonversi fungsi Python native apa pun menjadi objek lazy menggunakan dekorator @dask.delayed.

Berikut contoh sederhana:

%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s

Dalam contoh ini, kita mendefinisikan fungsi process_data yang didekorasi dengan @dask.delayed. Fungsi ini mensimulasikan pekerjaan komputasi dengan tidur selama 1 detik lalu mengembalikan kuadrat dari nilai masukan.

Tanpa paralelisme, melakukan komputasi ini pada 1000 masukan akan memakan waktu lebih dari 1000 detik. Namun, dengan Dask Delayed dan eksekusi paralel, komputasi hanya memakan waktu sekitar 42,1 detik.

Contoh ini menunjukkan kekuatan paralelisme dalam mengurangi waktu komputasi dengan mendistribusikan beban kerja secara efisien ke banyak inti atau worker.

Itulah esensi paralelisme.

Kesimpulan dan Sumber Lanjutan

Dask adalah salah satu pustaka pilar dalam ekosistem data. Dask memperluas fungsionalitas sebagian besar pustaka favorit seperti NumPy, Pandas, dan Spark, sehingga dapat menangani dataset yang lebih besar dari memori secara mulus.

Dengan Dask Bag dan Dask Delayed, Dask menghadirkan paralelisme dan lazy evaluation ke skenario nontradisional seperti bekerja dengan data tidak terstruktur atau objek Python native.

Untuk detail teknisnya, pastikan Anda membaca dengan saksama dokumentasi Dask.

Jika Anda mencari sumber komprehensif untuk menguasai Dask, lihat kursus kami,  Parallel Programming With Dask in Python.