Dask

Dask, mevcut Python ve PyData ekosistemini ölçeklendiren paralel ve dağıtık hesaplama kütüphanesidir.

Özellikle, pandas iş akışlarını ölçeklendirmek için Dask DataFrame kullanabiliriz. Dask DataFrame, büyük tabular verileri işlemek için pandas'ı paralelleştirir. pandas API'sine yakın bir şekilde yansır; bu da tek bir veri setinde test etmeden tam veri setini işlemeye geçişi kolaylaştırır. Dask, Hugging Face Datasets'in varsayılan biçimi olan Parquet ile özellikle etkilidir; çünkü zengin veri türlerini, verimli sütun bazlı filtrelemeyi ve sıkıştırmayı destekler.

Dask için iyi bir pratik kullanım senaryosu, veri işleme veya model çıkarımını bir veri setinde dağıtık bir şekilde çalıştırmaktır. Örneğin, Coiled’ın Hugging Face + Dask ile Yapay Zekâ Tabanlı Veri İşlemeyi Ölçeklendirme konusundaki mükemmel blog yazısına bakın.

Okuma ve Yazma

Dask, uzak verileri okumak ve yazmak için fsspec kullandığından, Hub'daki verileri okumak ve yazmak için Hugging Face yollarını (hf://) kullanabilirsiniz.

Önce Hugging Face hesabınızla oturum açmanız gerekir, örneğin şunu kullanarak:

hf auth login

Ardından bir veri seti deposu oluşturabilirsiniz, örneğin şunu kullanarak:

from huggingface_hub import HfApi

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

Son olarak, Dask'ta Hugging Face yollarını kullanabilirsiniz. Dask DataFrame, Hugging Face üzerinde Parquet'e dağıtık yazmayı destekler; bu, veri seti değişikliklerini izlemek için commit'leri kullanır:

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")

Bu işlem dosya başına bir commit oluşturduğundan, yükleme sonrasında geçmişi birleştirmek (squash) önerilir:

from huggingface_hub import HfApi

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

Bu, Parquet biçimindeki Dask veri setinizi içeren username/my_dataset adlı bir veri seti deposu oluşturur. Daha sonra yeniden yükleyebilirsiniz:

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")

Hugging Face yolları ve bunların nasıl uygulandığına dair daha fazla bilgi için lütfen HfFileSystem hakkında istemci kütüphanesinin belgelerine başvurun.

Verileri işleyin

Dask kullanarak bir veri setini paralel olarak işlemek için, önce bir pandas DataFrame veya Series için veri işleme fonksiyonunuzu tanımlayabilir, ardından bu fonksiyonu veri setinin tüm bölümlerine paralel olarak uygulamak için Dask map_partitions fonksiyonunu kullanabilirsiniz:

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

veya pandas string yöntemlerini kullanan benzer bir işlev (daha hızlı):

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

pandas'ta bu fonksiyonu bir metin sütununda kullanabilirsiniz:

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

Ve Dask'ta bu fonksiyonu her bölmede çalıştırabilirsiniz:

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

Çıktıdaki pandas Series veya DataFrame türü olan meta bilgisini de sağlamanız gerektiğini unutmayın. Bu, Dask DataFrame'in tembel (lazy) bir API kullandığı için gereklidir. Dask, veri işleme işlemini yalnızca .compute() çağrıldığında çalıştırdığından, bu arada yeni sütunun türünü bilmek için meta argümanına ihtiyaç duyar.

Öngörü ve Projeksiyon İtme

Hugging Face’ten Parquet verileri okurken Dask, Parquet dosyalarındaki metadata’yı otomatik olarak kullanarak gerekmediği durumlarda tüm dosyaları veya satır gruplarını atlar. Örneğin, Parquet biçimindeki bir Hugging Face Dataset üzerinde bir filtre (predikat) uygular veya sütunların bir alt kümesini seçerseniz (projeksiyon), Dask Parquet dosyalarının metadata’sını okuyarak indirmeden atılacak kısımları eler.

Bu, sorgu optimizasyonunu desteklemek için Dask DataFrame API'sinin yeniden implementasyonu sayesinde mümkündür; bu sayede Dask daha hızlı ve daha sağlam hale gelir.

Örneğin FineWeb-Edu'nun bu alt kümesi birçok Parquet dosyası içerir. Veri setini son CC dökümlerinden gelen metinleri koruyacak şekilde filtreleyebiliyorsanız, Dask dosyaların çoğunu atlar ve yalnızca filtreye uyan verileri indirir:

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, hesaplamanız için yalnızca gerekli sütunları okur ve diğerlerini atlar. Örneğin, kodunuzun ilerleyen bir kısmında bir sütunu düşürürseniz, boru hattının başlangıcında gerekmiyorsa onu yüklemekle uğraşmaz. Bu, sütunların bir alt kümesini işlemek veya analitik yapmak istediğinizde faydalıdır:

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

İstemci

dask içindeki özelliklerin çoğu, paralel hesaplamaları başlatmak için bir küme veya yerel bir Client için optimize edilmiştir:

import dask.dataframe as dd
from distributed import Client

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

Yerel kullanım için Client, varsayılan olarak çok işlemcili (multiprocessing) bir Dask LocalCluster kullanır. LocalCluster'in çok işlemciliğini aşağıdaki şekilde manuel olarak yapılandırabilirsiniz:

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

Yerel olarak Client olmadan varsayılan iş parçacıklı zamanlayıcıyı (threaded scheduler) kullanıyorsanız, belirli işlemlerden sonra bir DataFrame'in yavaşlayabileceğini unutmayın (daha fazla ayrıntı burada).

Yerel veya bulut kümesi kurulumuyla ilgili daha fazla bilgi için Dask dağıtım belgelerine göz atın.