Dask

Dask هو مكتبة للحوسبة المتوازية والموزعة التي توسّع نطاق منظومة Python وPyData الحالية.

على وجه الخصوص، يمكننا استخدام Dask DataFrame لتوسيع نطاق سير عمل pandas. يقوم Dask DataFrame بتوازي pandas للتعامل مع البيانات الجدولية الكبيرة. وهو يعكس واجهة برمجة تطبيقات pandas عن كثب، مما يجعل الانتقال من الاختبار على مجموعة بيانات واحدة إلى معالجة مجموعة البيانات الكاملة أمرًا بسيطًا. يُعد Dask فعالًا بشكل خاص مع Parquet، وهو التنسيق الافتراضي في Hugging Face Datasets، حيث يدعم أنواع البيانات الغنية، والتصفية العمودية الفعالة، والضغط.

من الاستخدامات العملية الجيدة لـ Dask هو تنفيذ معالجة البيانات أو استنتاج النموذج على مجموعة بيانات بطريقة موزعة. راجع، على سبيل المثال، المقال الرائع الذي نشرته Coiled حول Scaling AI-Based Data Processing with Hugging Face + Dask.

القراءة والكتابة

بما أن Dask يستخدم fsspec لقراءة البيانات عن بُعد وكتابتها، يمكنك استخدام مسارات Hugging Face (hf://) لقراءة البيانات على Hub وكتابتها.

أولاً، تحتاج إلى تسجيل الدخول بحساب Hugging Face الخاص بك، على سبيل المثال باستخدام:

hf auth login

ثم يمكنك إنشاء مستودع مجموعة بيانات، على سبيل المثال باستخدام:

from huggingface_hub import HfApi

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

أخيرًا، يمكنك استخدام مسارات Hugging Face في Dask. يدعم Dask DataFrame الكتابة الموزعة إلى Parquet على Hugging Face، والتي تستخدم الالتزامات (commits) لتتبع تغييرات مجموعة البيانات:

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

بما أن هذا يُنشئ التزامًا واحدًا لكل ملف، يُوصى بدمج السجل بعد الرفع:

from huggingface_hub import HfApi

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

ينشئ هذا مستودع بيانات username/my_dataset يحتوي على مجموعة بيانات Dask الخاصة بك بتنسيق Parquet. يمكنك إعادة تحميلها لاحقًا:

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 وكيفية تنفيذها، يُرجى الرجوع إلى توثيق مكتبة العميل الخاصة بـ HfFileSystem.

معالجة البيانات

لمعالجة مجموعة بيانات بالتوازي باستخدام Dask، يمكنك أولاً تعريف دالة معالجة البيانات الخاصة بك لـ pandas DataFrame أو Series، ثم استخدام دالة Dask map_partitions لتطبيق هذه الدالة على جميع أقسام مجموعة البيانات بالتوازي:

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

أو دالة مشابهة باستخدام طرق pandas النصية (أسرع):

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

في pandas يمكنك استخدام هذه الدالة على عمود نصي:

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

وفي Dask يمكنك تشغيل هذه الدالة على كل قسم:

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

لاحظ أنك بحاجة أيضًا إلى تقديم meta وهو نوع سلسلة pandas أو إطار البيانات في مخرجات دالتك. هذا ضروري لأن إطار بيانات Dask يستخدم واجهة برمجة تطبيقات كسولة. بما أن Dask لن ينفذ معالجة البيانات إلا عند استدعاء .compute()، فهو يحتاج إلى الحجة meta لمعرفة نوع العمود الجديد في غضون ذلك.

دفع التنبؤات والإسقاط

عند قراءة بيانات Parquet من Hugging Face، يستفيد Dask تلقائيًا من البيانات الوصفية (metadata) الموجودة في ملفات Parquet لتخطي الملفات أو مجموعات الصفوف (row groups) بالكامل إذا لم تكن مطلوبة. على سبيل المثال، إذا قمت بتطبيق عامل تصفية (predicate) على مجموعة بيانات Hugging Face بصيغة Parquet، أو إذا اخترت مجموعة فرعية من الأعمدة (projection)، فسيقوم Dask بقراءة البيانات الوصفية لملفات Parquet للتخلص من الأجزاء غير المطلوبة دون تحميلها.

يُعدّ ذلك ممكناً بفضل إعادة تنفيذ واجهة برمجة تطبيقات Dask DataFrame لدعم تحسين الاستعلامات، مما يجعل Dask أسرع وأكثر متانة.

على سبيل المثال، يحتوي هذا الجزء الفرعي من FineWeb-Edu على العديد من ملفات Parquet. إذا كان بإمكانك تصفية مجموعة البيانات للاحتفاظ بالنصوص من نسخ CC الحديثة، فسيتم تجاهل معظم الملفات بواسطة Dask وتحميل البيانات المطابقة للتصفية فقط:

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 أيضًا الأعمدة المطلوبة فقط لحسابك ويتخطى الباقي. فعلى سبيل المثال، إذا قمت بحذف عمود في مرحلة متأخرة من كودك، فلن يُحمّله في المراحل المبكرة من خط المعالجة إذا لم يكن مطلوبًا. يكون هذا مفيدًا عندما تريد التعامل مع مجموعة فرعية من الأعمدة أو لأغراض التحليلات:

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

العميل

معظم الميزات في dask مُحسَّنة لمجموعة (cluster) أو Client محلي لإطلاق الحسابات المتوازية:

import dask.dataframe as dd
from distributed import Client

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

للاستخدام المحلي، يستخدم Client عنصراً من Dask يُسمى LocalCluster مع تعدد العمليات (multiprocessing) افتراضياً. يمكنك ضبط تعدد العمليات في LocalCluster يدوياً باستخدام

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

لاحظ أنه إذا كنت تستخدم جدولة الخيوط الافتراضية محليًا دون Client، فقد يصبح DataFrame أبطأ بعد عمليات معينة (المزيد من التفاصيل هنا).

يمكنك العثور على مزيد من المعلومات حول إعداد عناقيد محلية أو سحابية في توثيق نشر Dask.