Dask

Dask adalah pustaka komputasi paralel dan terdistribusi yang memperluas ekosistem Python dan PyData yang ada.

Secara khusus, kita dapat menggunakan Dask DataFrame untuk memperluas skala alur kerja pandas. Dask DataFrame memparalelkan pandas untuk menangani data tabular besar. API-nya sangat mirip dengan API pandas, sehingga transisi dari pengujian pada satu set data ke pemrosesan set data penuh menjadi sederhana. Dask sangat efektif dengan Parquet, format default pada Hugging Face Datasets, karena mendukung tipe data yang kaya, penyaringan kolom yang efisien, dan kompresi.

Kasus penggunaan praktis yang baik untuk Dask adalah menjalankan pemrosesan data atau inferensi model pada dataset secara terdistribusi. Lihat, misalnya, artikel blog yang sangat baik dari Coiled tentang Scaling AI-Based Data Processing with Hugging Face + Dask.

Baca dan Tulis

Karena Dask menggunakan fsspec untuk membaca dan menulis data jarak jauh, Anda dapat menggunakan jalur Hugging Face (hf://) untuk membaca dan menulis data di Hub.

Pertama, Anda perlu Masuk dengan akun Hugging Face Anda, misalnya menggunakan:

hf auth login

Lalu Anda dapat Membuat repositori dataset, misalnya menggunakan:

from huggingface_hub import HfApi

HfApi().create_repo(repo_id="username/my_dataset", repo_type="dataset")

Terakhir, Anda dapat menggunakan jalur Hugging Face di Dask. Dask DataFrame mendukung penulisan terdistribusi ke Parquet di Hugging Face, yang menggunakan commit untuk melacak perubahan dataset:

import dask.dataframe as dd

df.to_parquet("hf://datasets/username/my_dataset")

# or write in separate directories if the dataset has train/validation/test splits
df_train.to_parquet("hf://datasets/username/my_dataset/train")
df_valid.to_parquet("hf://datasets/username/my_dataset/validation")
df_test .to_parquet("hf://datasets/username/my_dataset/test")

Karena hal ini membuat satu commit per file, disarankan untuk melakukan squash pada riwayat setelah pengunggahan:

from huggingface_hub import HfApi

HfApi().super_squash_history(repo_id=repo_id, repo_type="dataset")

Ini membuat repositori dataset username/my_dataset yang berisi dataset Dask Anda dalam format Parquet. Anda dapat memuatnya kembali nanti:

import dask.dataframe as dd

df = dd.read_parquet("hf://datasets/username/my_dataset")

# or read from separate directories if the dataset has train/validation/test splits
df_train = dd.read_parquet("hf://datasets/username/my_dataset/train")
df_valid = dd.read_parquet("hf://datasets/username/my_dataset/validation")
df_test  = dd.read_parquet("hf://datasets/username/my_dataset/test")

Untuk informasi lebih lanjut tentang jalur Hugging Face dan cara implementasinya, silakan merujuk ke dokumentasi pustaka klien tentang HfFileSystem.

Proses data

Untuk memproses dataset secara paralel menggunakan Dask, Anda dapat terlebih dahulu mendefinisikan fungsi pemrosesan data Anda untuk pandas DataFrame atau Series, lalu menggunakan fungsi map_partitions Dask untuk menerapkan fungsi ini ke semua partisi dataset secara paralel:

def dummy_count_words(texts):
    return pd.Series([len(text.split(" ")) for text in texts])

atau fungsi serupa menggunakan metode string pandas (lebih cepat):

def dummy_count_words(texts):
    return texts.str.count(" ")

Di pandas, Anda dapat menggunakan fungsi ini pada kolom teks:

# pandas API
df["num_words"] = dummy_count_words(df.text)

Dan di Dask, Anda dapat menjalankan fungsi ini di setiap partisi:

# Dask API: run the function on every partition
df["num_words"] = df.text.map_partitions(dummy_count_words, meta=int)

Perhatikan bahwa Anda juga perlu menyediakan meta, yaitu tipe dari pandas Series atau DataFrame dalam output fungsi Anda. Hal ini diperlukan karena Dask DataFrame menggunakan API lazy. Karena Dask hanya akan menjalankan pemrosesan data setelah .compute() dipanggil, ia membutuhkan argumen meta untuk mengetahui tipe kolom baru di sela-sela waktu tersebut.

Predicate dan Projection Pushdown

Saat membaca data Parquet dari Hugging Face, Dask secara otomatis memanfaatkan metadata dalam file Parquet untuk melewati seluruh file atau grup baris jika tidak diperlukan. Misalnya, jika Anda menerapkan filter (predikat) pada Dataset Hugging Face dalam format Parquet atau jika Anda memilih subset kolom (proyeksi), Dask akan membaca metadata file Parquet untuk membuang bagian yang tidak diperlukan tanpa mengunduhnya.

Hal ini dimungkinkan berkat implementasi ulang API Dask DataFrame untuk mendukung optimasi kueri, yang membuat Dask lebih cepat dan lebih tangguh.

Sebagai contoh, subset FineWeb-Edu ini berisi banyak file Parquet. Jika Anda dapat memfilter dataset untuk mempertahankan teks dari dump CC terbaru, Dask akan melewati sebagian besar file dan hanya mengunduh data yang cocok dengan filter:

import dask.dataframe as dd

df = dd.read_parquet("hf://datasets/HuggingFaceFW/fineweb-edu/sample/10BT/*.parquet")

# Dask will skip the files or row groups that don't
# match the query without downloading them.
df = df[df.dump >= "CC-MAIN-2023"]

Dask juga hanya akan membaca kolom yang diperlukan untuk komputasi Anda dan melewatkan sisanya. Sebagai contoh, jika Anda menghapus kolom di bagian akhir kode, Dask tidak akan memuatnya di awal pipeline jika tidak diperlukan. Hal ini berguna ketika Anda ingin memanipulasi subset kolom atau untuk analitik:

# Dask will download the 'dump' and 'token_count' needed
# for the filtering and computation and skip the other columns.
df.token_count.mean().compute()

Klien

Sebagian besar fitur di dask dioptimalkan untuk klaster atau Client lokal untuk meluncurkan komputasi paralel:

import dask.dataframe as dd
from distributed import Client

if __name__ == "__main__":  # needed for creating new processes
    client = Client()
    df = dd.read_parquet(...)
    ...

Untuk penggunaan lokal, Client menggunakan Dask LocalCluster dengan multiprocessing secara default. Anda dapat mengonfigurasi multiprocessing dari LocalCluster secara manual dengan

from dask.distributed import Client, LocalCluster
cluster = LocalCluster(n_workers=8, threads_per_worker=8)
client = Client(cluster)

Perhatikan bahwa jika Anda menggunakan scheduler berulir default secara lokal tanpa Client, DataFrame dapat menjadi lebih lambat setelah operasi tertentu (detail lebih lanjut di sini).

Temukan informasi lebih lanjut tentang pengaturan klaster lokal atau cloud dalam dokumentasi Deploying Dask.