توسيع نطاق معالجة البيانات القائمة على الذكاء الاصطناعي من خلال Hugging Face + Dask

تحتوي منصة Hugging Face على العديد من مجموعات البيانات والنماذج المدربة مسبقًا والتي تجعل استخدام أحدث نماذج التعلم الآلي وتدريبها متاحًا بشكل متزايد. ومع ذلك، قد يكون من الصعب توسيع نطاق مهام الذكاء الاصطناعي لأن مجموعات بيانات الذكاء الاصطناعي غالبًا ما تكون كبيرة (100 جيجا بايت إلى تيرابايت) واستخدام محولات Hugging Face لاستدلال النموذج قد يكون مكلفًا من الناحية الحسابية في بعض الأحيان.

يمكن لمكتبة Dask، وهي مكتبة بايثون للحوسبة الموزعة، التعامل مع الحوسبة خارج النواة (معالجة البيانات التي لا تتناسب مع الذاكرة) عن طريق تقسيم مجموعات البيانات إلى أجزاء يمكن التحكم فيها. وهذا يجعل من السهل القيام بأشياء مثل:

• تحميل البيانات بكفاءة ومعالجتها مسبقًا لمجموعات البيانات على نطاق السل مع واجهة برمجة تطبيقات سهلة الاستخدام تحاكي حيوانات الباندا
• استنتاج النموذج الموازي (مع خيار استنتاج GPU متعدد العقد)

نعرض في هذا المنشور مثالاً على معالجة البيانات من مجموعة بيانات FineWeb، باستخدام مصنف FineWeb-Edu لتحديد صفحات الويب ذات القيمة التعليمية العالية. سوف نعرض:

• كيفية معالجة 100 صف محليًا باستخدام الباندا
• التوسع إلى 211 مليون صف باستخدام Dask عبر وحدات معالجة الرسومات المتعددة على السحابة

معالجة 100 صف مع الباندا

تتكون مجموعة بيانات FineWeb من 15 تريليون رمز مميز لبيانات الويب باللغة الإنجليزية من Common Crawl، وهي مؤسسة غير ربحية تستضيف مجموعة بيانات عامة للزحف على الويب يتم تحديثها شهريًا. غالبًا ما تُستخدم مجموعة البيانات هذه لمجموعة متنوعة من المهام مثل التدريب على نماذج اللغة الكبيرة والتصنيف وتصفية المحتوى واسترجاع المعلومات عبر مجموعة متنوعة من القطاعات.

قد يستغرق الأمر أكثر من دقيقة واحدة للتنزيل والقراءة في ملف واحد باستخدام الباندا على جهاز كمبيوتر محمول.

import pandas as pd

df = pd.read_parquet(
    "hf://datasets/HuggingFaceFW/fineweb/data/CC-MAIN-2024-10/000_00000.parquet"
)

بعد ذلك، سنستخدم مصنف HF FineWeb-Edu للحكم على القيمة التعليمية لصفحات الويب في مجموعة البيانات الخاصة بنا. يتم ترتيب صفحات الويب على مقياس من 0 إلى 5، حيث يشير 0 إلى غير تعليمي و5 إلى تعليم عالٍ. يمكننا استخدام الباندا للقيام بذلك على مجموعة فرعية أصغر مكونة من 100 صف من البيانات، الأمر الذي يستغرق حوالي 10 ثوانٍ على جهاز M1 Mac المزود بوحدة معالجة الرسومات.

from transformers import pipeline

def compute_scores(texts):
    import torch

    
    if torch.cuda.is_available():
        device = torch.device("cuda")
    elif torch.backends.mps.is_available():
        device = torch.device("mps")
    else:
        device = torch.device("cpu")

    pipe = pipeline(
        "text-classification",
        model="HuggingFaceFW/fineweb-edu-classifier",
        device=device
    )
    results = pipe(
        texts.to_list(),
        batch_size=25,                    
        padding="longest",
        truncation=True,
        function_to_apply="none"
    )
    
    return pd.Series([r["score"] for r in results])

df = df[:100]
min_edu_score = 3
df["edu-classifier-score"] = compute_scores(df.text)
df = df[df["edu-classifier-score"] >= min_edu_score]

لاحظ أننا أضفنا أيضًا خطوة للتحقق من الأجهزة المتوفرة داخل الملف compute_scores وظيفة، لأنه سيتم توزيعها عندما نتوسع باستخدام Dask في الخطوة التالية. وهذا يجعل من السهل الانتقال من الاختبار محليًا على جهاز واحد (إما على وحدة المعالجة المركزية أو ربما لديك جهاز MacBook مزود بوحدة معالجة رسومات Apple silicon) إلى التوزيع عبر أجهزة متعددة (مثل وحدات معالجة الرسومات NVIDIA).

التوسع إلى 211 مليون صف باستخدام Dask

تبلغ مساحة الزحف الكاملة لشهر فبراير/مارس 2024 432 جيجابايت على القرص، أو 715 جيجابايت تقريبًا في الذاكرة، مقسمة على 250 ملف باركيه. حتى على جهاز به ذاكرة كافية لمجموعة البيانات بأكملها، سيكون تنفيذ ذلك بطيئًا للغاية بشكل تسلسلي.

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

import dask.dataframe as dd

df = dd.read_parquet(
    
    "hf://datasets/HuggingFaceFW/fineweb/data/CC-MAIN-2024-10/*.parquet" 
)

سوف نقوم بتطبيق compute_scores وظيفة لتصنيف النص بالتوازي على Dask DataFrame الخاص بنا باستخدام map_partitions، والذي يطبق وظيفتنا بالتوازي على كل Pandas DataFrame في Dask DataFrame الأكبر. ال meta الوسيطة خاصة بـ Dask، وتشير إلى بنية البيانات (أسماء الأعمدة وأنواع البيانات) للمخرجات.

from transformers import pipeline

def compute_scores(texts):
    import torch

    
    if torch.cuda.is_available():
        device = torch.device("cuda")
    elif torch.backends.mps.is_available():
        device = torch.device("mps")
    else:
        device = torch.device("cpu")

    pipe = pipeline(
        "text-classification",
        model="HuggingFaceFW/fineweb-edu-classifier",
        device=device,
    )
    results = pipe(
        texts.to_list(),
        batch_size=768,
        padding="longest",
        truncation=True,
        function_to_apply="none",
    )

    return pd.Series([r["score"] for r in results])

min_edu_score = 3
df["edu-classifier-score"] = df.text.map_partitions(compute_scores, meta=pd.Series([0]))
df = df[df["edu-classifier-score"] >= min_edu_score]

لاحظ أننا اخترنا أ batch_size يعمل بشكل جيد مع هذا المثال، ولكن من المحتمل أن ترغب في تخصيص ذلك اعتمادًا على الأجهزة والبيانات والنموذج الذي تستخدمه في سير العمل الخاص بك (راجع مستندات HF حول تجميع خطوط الأنابيب).

الآن بعد أن حددنا صفوف مجموعة البيانات التي نهتم بها، يمكننا حفظ النتيجة لإجراء تحليلات أخرى. يدعم Dask DataFrame الكتابة الموزعة على Parquet تلقائيًا. يستخدم Hugging Face الالتزامات لتتبع تغييرات مجموعة البيانات ويسمح بكتابة Dask DataFrame بالتوازي.

repo_id = "<your-hf-user>/<your-dataset-name>"  
df.to_parquet(f"hf://datasets/{repo_id}")

نظرًا لأن هذا يؤدي إلى إنشاء التزام واحد لكل ملف، فمن المستحسن سحق السجل بعد التحميل:

from huggingface_hub import HfApi

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

وبدلاً من ذلك، يمكنك استخدام هذه الوظيفة المخصصة التي تقوم بتحميل ملفات متعددة لكل التزام.

استنتاج النموذج الموازي لوحدة معالجة الرسومات المتعددة

هناك عدة طرق لنشر Dask على مجموعة متنوعة من الأجهزة. سنستخدم هنا Coiled لنشر Dask على السحابة حتى نتمكن من تشغيل الأجهزة الافتراضية حسب الحاجة، ثم تنظيفها عند الانتهاء.

cluster = coiled.Cluster(
    region="us-east-1",                 
    n_workers=100,                      
    spot_policy="spot_with_fallback",   
    worker_vm_types="g5.xlarge",        
    worker_options={"nthreads": 1},
)
client = cluster.get_client()

تحت غطاء محرك السيارة مقابض ملفوفة:

• توفير أجهزة افتراضية سحابية مزودة بأجهزة GPU. في هذه الحالة، g5.xlarge مثيلات على AWS.
• إعداد برامج تشغيل NVIDIA المناسبة، ووقت تشغيل CUDA، وما إلى ذلك.
• التثبيت التلقائي لنفس الحزم الموجودة لديك محليًا على السحابة الافتراضية مع مزامنة الحزمة. يتضمن ذلك ملفات بايثون في دليل العمل الخاص بك.

استغرق سير العمل حوالي 5 ساعات حتى يكتمل وكان لدينا استخدام جيد لأجهزة GPU.

بتجميع كل ذلك معًا، إليك سير العمل الكامل:

import dask.dataframe as dd
from transformers import pipeline
from huggingface_hub import HfApi
import os
import coiled

cluster = coiled.Cluster(
    region="us-east-1",
    n_workers=100,
    spot_policy="spot_with_fallback",
    worker_vm_types="g5.xlarge",
    worker_options={"nthreads": 1},
)
client = cluster.get_client()
cluster.send_private_envs(
    {"HF_TOKEN": "<your-hf-token>"}             
)

df = dd.read_parquet(
    "hf://datasets/HuggingFaceFW/fineweb/data/CC-MAIN-2024-10/*.parquet"
)

def compute_scores(texts):
    import torch

    
    if torch.cuda.is_available():
        device = torch.device("cuda")
    elif torch.backends.mps.is_available():
        device = torch.device("mps")
    else:
        device = torch.device("cpu")

    pipe = pipeline(
        "text-classification",
        model="HuggingFaceFW/fineweb-edu-classifier",
        device=device
    )
    results = pipe(
        texts.to_list(),
        batch_size=768,
        padding="longest",
        truncation=True,
        function_to_apply="none"
    )

    return pd.Series([r["score"] for r in results])

min_edu_score = 3
df["edu-classifier-score"] = df.text.map_partitions(compute_scores, meta=pd.Series([0]))
df = df[df["edu-classifier-score"] >= min_edu_score]

repo_id = "<your-hf-user>/<your-dataset-name>"  
df.to_parquet(f"hf://datasets/{repo_id}")

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

خاتمة

يعتبر Hugging Face + Dask مزيجًا قويًا. في هذا المثال، قمنا بتوسيع نطاق مهمة التصنيف لدينا من 100 صف إلى 211 مليون صف باستخدام Dask + Coiled لتشغيل سير العمل بالتوازي عبر وحدات معالجة الرسومات المتعددة على السحابة.

يمكن استخدام نفس النوع من سير العمل في حالات الاستخدام الأخرى مثل:

• تصفية البيانات الجينومية لتحديد الجينات ذات الاهتمام
• استخراج المعلومات من النص غير المنظم وتحويلها إلى مجموعات بيانات منظمة
• تنظيف البيانات النصية المستخرجة من الإنترنت أو الزحف المشترك
• تشغيل استدلال نموذج متعدد الوسائط لتحليل مجموعات بيانات الصوت أو الصورة أو الفيديو الكبيرة