Pandera دليل التحقق من البيانات (2026)

مقدمة: ليش التحقق من صحة البيانات مهم جدًا؟

إذا كنت تشتغل في مجال علوم البيانات أو هندسة البيانات، فأنت بالتأكيد سمعت الإحصائية الشهيرة: محللو البيانات يقضون ما يصل إلى 80% من وقتهم في تنظيف البيانات وإعدادها. لكن المشكلة الأعمق — وهذي اللي كثير ناس يتجاهلونها — مو بس في تنظيف البيانات، بل في التأكد إن البيانات اللي تمر عبر أنابيب المعالجة (pipelines) صحيحة فعلًا ومطابقة للتوقعات.

القاعدة الذهبية واضحة: "قمامة تدخل، قمامة تخرج" (Garbage In, Garbage Out).

مهما كانت خوارزمياتك متقدمة أو نماذجك معقدة، النتائج بتكون عديمة القيمة إذا كانت البيانات المدخلة فاسدة أو غير متسقة. صراحةً، شفت هالمشكلة بنفسي أكثر من مرة في مشاريع حقيقية.

تخيل هالسيناريو الشائع: تبني نموذج تعلم آلي للتنبؤ بالمبيعات. يشتغل النموذج بشكل ممتاز في بيئة التطوير. بعدين يُنشر في بيئة الإنتاج ويبدأ يعطي تنبؤات خاطئة تمامًا. بعد ساعات من التحقيق، تكتشف إن مصدر البيانات بدأ يرسل أسعارًا سالبة، أو إن عمود التاريخ تغير تنسيقه، أو إن فئة جديدة ظهرت في عمود كان يحتوي على ثلاث فئات فقط. كل هالمشاكل كان يمكن اكتشافها فورًا لو كان عندك طبقة تحقق من صحة البيانات.

التحقق اليدوي باستخدام عبارات assert وشروط if المتناثرة في الكود؟ مو حل قابل للتوسع. هالنهج هش وصعب الصيانة ولا يوفر رسائل خطأ واضحة. هنا يجي دور مكتبة Pandera — الأداة المصممة خصيصًا للتحقق من صحة البيانات في إطارات بيانات pandas بطريقة تصريحية، قابلة للقراءة، وجاهزة للإنتاج.

ما هي مكتبة Pandera ولماذا تستخدمها؟

Pandera هي مكتبة بايثون مفتوحة المصدر للتحقق الإحصائي من صحة بيانات إطارات البيانات (DataFrames). الإصدار الأحدث هو 0.29.0 اللي صدر في يناير 2026، ويتطلب Python 3.10 أو أحدث.

الشي اللي يميز Pandera فعلًا هو إنها مو مخصصة لـ pandas بس. تدعم أيضًا Polars وDask وModin وIbis وPySpark، وهالشي يخليها أداة موحدة للتحقق من البيانات بغض النظر عن محرك المعالجة اللي تستخدمه.

طيب، ليش تستخدم Pandera بدل التحقق اليدوي؟ إليك الأسباب الرئيسية:

تصريحية وواضحة: تعريف المخطط (schema) يوثّق توقعاتك حول البيانات بشكل صريح وقابل للقراءة
رسائل خطأ مفصلة: عند فشل التحقق، تحصل على تقرير دقيق يحدد أي عمود، أي صف، وأي قاعدة تم انتهاكها
التحقق الكسول (Lazy Validation): إمكانية جمع جميع الأخطاء دفعة واحدة بدل ما يتوقف عند أول خطأ
مُزخرفات الدوال (Decorators): التحقق التلقائي من مدخلات ومخرجات الدوال في أنابيب المعالجة
فحوصات إحصائية: التحقق من التوزيعات الإحصائية واختبار الفرضيات مباشرة في المخطط
قابلة للتكامل مع CI/CD: إضافة التحقق من البيانات كخطوة في خط أنابيب النشر المستمر

تثبيت Pandera

تثبيت Pandera بسيط ومباشر. الأفضل تثبيتها مع الدعم الكامل لـ pandas:

# تثبيت Pandera مع دعم pandas
pip install 'pandera[pandas]'

# أو تثبيت الحد الأدنى فقط
pip install pandera

# للتحقق من الإصدار المثبت
python -c "import pandera; print(pandera.__version__)"
# 0.29.0

إذا كنت تستخدم محركات معالجة أخرى، تقدر تثبت الدعم المناسب:

# لدعم Polars
pip install 'pandera[polars]'

# لدعم Dask
pip install 'pandera[dask]'

# لدعم PySpark
pip install 'pandera[pyspark]'

# لتثبيت جميع المحركات المدعومة
pip install 'pandera[all]'

تعريف المخططات باستخدام DataFrameSchema

الطريقة الأولى لتعريف مخطط التحقق في Pandera هي استخدام DataFrameSchema بأسلوب القاموس (dictionary-style). هالطريقة مرنة ومناسبة للمخططات الديناميكية اللي ممكن تُبنى برمجيًا أثناء وقت التشغيل.

import pandas as pd
import pandera as pa

# إنشاء بيانات نموذجية لمبيعات متجر إلكتروني
df = pd.DataFrame({
    "order_id": [1001, 1002, 1003, 1004, 1005],
    "product_name": ["حاسوب محمول", "هاتف ذكي", "سماعات", "لوحة مفاتيح", "شاشة"],
    "price": [2500.00, 1200.00, 150.00, 80.00, 3500.00],
    "quantity": [1, 2, 5, 3, 1],
    "category": ["إلكترونيات", "إلكترونيات", "إكسسوارات", "إكسسوارات", "إلكترونيات"],
})

# تعريف مخطط التحقق باستخدام DataFrameSchema
schema = pa.DataFrameSchema(
    columns={
        "order_id": pa.Column(
            int,
            nullable=False,
            unique=True,
            description="معرّف الطلب الفريد"
        ),
        "product_name": pa.Column(
            str,
            checks=pa.Check.str_length(min_value=1, max_value=200),
            nullable=False,
            description="اسم المنتج"
        ),
        "price": pa.Column(
            float,
            checks=[
                pa.Check.greater_than(0, error="السعر يجب أن يكون أكبر من صفر"),
                pa.Check.less_than(1_000_000, error="السعر يتجاوز الحد الأقصى"),
            ],
            nullable=False,
            description="سعر المنتج بالريال"
        ),
        "quantity": pa.Column(
            int,
            checks=pa.Check.in_range(min_value=1, max_value=10000),
            nullable=False,
            description="الكمية المطلوبة"
        ),
        "category": pa.Column(
            str,
            checks=pa.Check.isin(["إلكترونيات", "إكسسوارات", "أجهزة منزلية"]),
            nullable=False,
            description="فئة المنتج"
        ),
    },
    strict=False,   # السماح بأعمدة إضافية غير معرّفة في المخطط
    coerce=True,    # تحويل الأنواع تلقائيًا إن أمكن
)

# تنفيذ التحقق
df_validated = schema.validate(df)
print("تم التحقق من صحة البيانات بنجاح!")
print(df_validated)

في المثال أعلاه، نعرّف مخططًا يحدد لكل عمود: نوع البيانات المتوقع، هل القيم الفارغة مسموحة، وفحوصات إضافية مثل النطاق المسموح وطول النص والقيم المقبولة. إذا البيانات ما استوفت أي من هالشروط، بتُثار استثناء SchemaError مع رسالة توضح المشكلة بدقة.

تعريف المخططات باستخدام DataFrameModel (الأسلوب المفضل)

الطريقة الثانية — والمفضلة صراحةً — هي استخدام DataFrameModel بأسلوب الفئات (class-based). إذا سبق لك استخدام Pydantic، بتحس إن الأسلوب مألوف جدًا. هالنهج أنظف وأسهل قراءة وصيانة، ويوفر تكاملًا أفضل مع أدوات تحليل الأنواع (type checkers).

import pandas as pd
import pandera as pa
from pandera.typing import Series

class SalesSchema(pa.DataFrameModel):
    """مخطط التحقق من بيانات المبيعات."""

    order_id: Series[int] = pa.Field(
        nullable=False,
        unique=True,
        ge=1000,                          # أكبر من أو يساوي 1000
        description="معرّف الطلب الفريد"
    )
    product_name: Series[str] = pa.Field(
        nullable=False,
        str_length={"min_value": 1, "max_value": 200},
        description="اسم المنتج"
    )
    price: Series[float] = pa.Field(
        gt=0,                              # أكبر من صفر تمامًا
        lt=1_000_000,                      # أصغر من مليون
        nullable=False,
        description="سعر المنتج"
    )
    quantity: Series[int] = pa.Field(
        ge=1,                              # أكبر من أو يساوي 1
        le=10000,                          # أصغر من أو يساوي 10000
        nullable=False,
        description="الكمية المطلوبة"
    )
    category: Series[str] = pa.Field(
        nullable=False,
        isin=["إلكترونيات", "إكسسوارات", "أجهزة منزلية"],
        description="فئة المنتج"
    )

    class Config:
        strict = False    # السماح بأعمدة إضافية
        coerce = True     # تحويل الأنواع تلقائيًا
        name = "مخطط المبيعات"

# إنشاء بيانات للاختبار
df = pd.DataFrame({
    "order_id": [1001, 1002, 1003, 1004],
    "product_name": ["حاسوب محمول", "هاتف ذكي", "سماعات", "شاشة"],
    "price": [2500.0, 1200.0, 150.0, 3500.0],
    "quantity": [1, 2, 5, 1],
    "category": ["إلكترونيات", "إلكترونيات", "إكسسوارات", "إلكترونيات"],
})

# التحقق باستخدام DataFrameModel
df_validated = SalesSchema.validate(df)
print("تم التحقق من صحة البيانات بنجاح!")
print(f"عدد الصفوف: {len(df_validated)}")

لاحظ كيف إن أسلوب DataFrameModel يشبه تعريف الفئات في Pydantic: كل حقل يُعرّف بتوصيف النوع (type annotation) مع Series[type]، والقيود تُحدد من خلال pa.Field(). الإعدادات العامة للمخطط تُحدد في الفئة الداخلية Config.

هالأسلوب هو المفضل في المشاريع الحديثة لأنه أكثر تنظيمًا وقابلية للصيانة. أنصح فيه بشدة لأي مشروع جديد.

الفحوصات المدمجة والفحوصات المخصصة

الفحوصات المدمجة

Pandera توفر مجموعة واسعة من الفحوصات المدمجة (Built-in Checks) اللي تغطي معظم سيناريوهات التحقق الشائعة. خلنا نشوف أهمها:

import pandas as pd
import pandera as pa

schema = pa.DataFrameSchema({
    # فحوصات رقمية
    "age": pa.Column(int, checks=[
        pa.Check.in_range(0, 150),           # النطاق: من 0 إلى 150
        pa.Check.greater_than_or_equal_to(0), # غير سالب
    ]),

    # فحوصات نصية
    "email": pa.Column(str, checks=[
        pa.Check.str_matches(r'^[\w\.-]+@[\w\.-]+\.\w+$'),  # نمط البريد الإلكتروني
        pa.Check.str_length(min_value=5, max_value=254),      # طول النص
    ]),

    # فحوصات القيم المحددة
    "status": pa.Column(str, checks=[
        pa.Check.isin(["active", "inactive", "suspended"]),  # قيم مسموحة فقط
    ]),

    # فحوصات النص مع التعبيرات النمطية
    "phone": pa.Column(str, checks=[
        pa.Check.str_startswith("+"),          # يبدأ بعلامة +
        pa.Check.str_length(min_value=10, max_value=15),
    ]),

    # فحوصات القيم الفريدة والفارغة
    "user_id": pa.Column(int, nullable=False, unique=True),
})

الفحوصات المخصصة

لما الفحوصات المدمجة ما تكفي (وصدقني، بيجيك هاليوم)، تقدر تنشئ فحوصات مخصصة بسهولة باستخدام pa.Check مع دالة لامبدا أو دالة عادية:

import pandas as pd
import pandera as pa
import numpy as np

# فحص مخصص باستخدام دالة لامبدا (على مستوى العنصر)
check_even = pa.Check(lambda x: x % 2 == 0, element_wise=True,
                       error="يجب أن تكون القيمة زوجية")

# فحص مخصص على مستوى السلسلة الكاملة (Series-level)
check_mean_range = pa.Check(
    lambda series: 100 <= series.mean() <= 500,
    error="المتوسط يجب أن يكون بين 100 و 500"
)

# فحص مخصص بدالة كاملة لمنطق أعقد
def check_no_outliers(series: pd.Series) -> bool:
    """التحقق من عدم وجود قيم شاذة باستخدام طريقة IQR."""
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 3 * IQR
    upper_bound = Q3 + 3 * IQR
    return series.between(lower_bound, upper_bound).all()

# استخدام الفحوصات المخصصة في المخطط
schema = pa.DataFrameSchema({
    "price": pa.Column(float, checks=[
        pa.Check.greater_than(0),
        check_mean_range,
        pa.Check(check_no_outliers, error="تم اكتشاف قيم شاذة في الأسعار"),
    ]),
    "quantity": pa.Column(int, checks=[
        pa.Check.greater_than(0),
        check_even,
    ]),
})

# فحوصات مخصصة في DataFrameModel باستخدام المزخرفات
class ProductSchema(pa.DataFrameModel):
    price: pa.typing.Series[float] = pa.Field(gt=0)
    cost: pa.typing.Series[float] = pa.Field(gt=0)

    # فحص مخصص على مستوى العمود
    @pa.check("price")
    def price_is_reasonable(cls, series: pd.Series) -> bool:
        """التحقق من أن الأسعار ضمن نطاق معقول."""
        return series.between(0.01, 999999.99).all()

    # فحص مخصص على مستوى DataFrame بالكامل
    @pa.dataframe_check
    def price_greater_than_cost(cls, df: pd.DataFrame) -> bool:
        """التحقق من أن السعر أكبر من التكلفة دائمًا."""
        return (df["price"] > df["cost"]).all()

الفحوصات المخصصة هي اللي تعطي Pandera قوتها الحقيقية. تقدر تكتب أي منطق تحقق تحتاجه — من فحوصات بسيطة مثل التحقق من الأرقام الزوجية، إلى فحوصات معقدة مثل كشف القيم الشاذة باستخدام طريقة IQR.

التحقق الكسول — جمع جميع الأخطاء دفعة واحدة

بشكل افتراضي، Pandera تتوقف عند أول خطأ تكتشفه. هالسلوك مفيد في بيئة الإنتاج (الفشل السريع أو fail-fast)، لكنه صراحةً غير عملي أبدًا أثناء التطوير وتنقيح الأخطاء.

التحقق الكسول (lazy=True) يجمع جميع الأخطاء ويعيدها في تقرير شامل واحد. وهالشي يوفر عليك وقت كبير.

import pandas as pd
import pandera as pa

# بيانات تحتوي على عدة مشاكل متزامنة
df_problematic = pd.DataFrame({
    "order_id": [1001, 1002, 1002, 1004, 1005],  # معرّف مكرر (1002)
    "product_name": ["حاسوب", "", None, "هاتف", "شاشة"],  # نص فارغ وقيمة مفقودة
    "price": [2500.0, -100.0, 150.0, 0.0, 3500.0],  # سعر سالب وسعر صفري
    "quantity": [1, 2, -3, 1, 0],  # كمية سالبة وكمية صفرية
})

schema = pa.DataFrameSchema({
    "order_id": pa.Column(int, nullable=False, unique=True),
    "product_name": pa.Column(
        str,
        checks=pa.Check.str_length(min_value=1),
        nullable=False
    ),
    "price": pa.Column(
        float,
        checks=pa.Check.greater_than(0),
        nullable=False
    ),
    "quantity": pa.Column(
        int,
        checks=pa.Check.greater_than(0),
        nullable=False
    ),
}, coerce=True)

# التحقق الكسول: جمع جميع الأخطاء
try:
    schema.validate(df_problematic, lazy=True)
except pa.errors.SchemaErrors as exc:
    print("=== تقرير أخطاء التحقق ===")
    print(f"عدد الأخطاء المكتشفة: {len(exc.failure_cases)}")
    print()
    # عرض تفاصيل كل خطأ
    print(exc.failure_cases.to_string())
    # يعرض: اسم العمود، نوع الفحص الفاشل، القيم المسببة للمشكلة، أرقام الصفوف
    print()
    # يمكنك أيضًا الوصول إلى رسالة الخطأ الكاملة
    print(exc.message)

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

مُزخرفات الدوال للتكامل مع أنابيب المعالجة

هالجزء من أقوى ميزات Pandera بالنسبة لي. مُزخرفات الدوال (Function Decorators) تتحقق تلقائيًا من مدخلات ومخرجات دوال المعالجة. باستخدام check_input وcheck_output وcheck_io، تقدر تضمن إن كل دالة في أنبوب المعالجة تستقبل بيانات صحيحة وتُخرج بيانات صحيحة.

import pandas as pd
import pandera as pa
from pandera import check_input, check_output, check_io
from pandera.typing import Series, DataFrame

# تعريف مخطط البيانات الخام (المدخلات)
class RawSalesSchema(pa.DataFrameModel):
    """مخطط البيانات الخام قبل المعالجة."""
    order_id: Series[int] = pa.Field(nullable=False)
    product_name: Series[str] = pa.Field(nullable=False)
    price: Series[float] = pa.Field(nullable=False)
    quantity: Series[int] = pa.Field(ge=1, nullable=False)

    class Config:
        strict = False
        coerce = True

# تعريف مخطط البيانات المعالجة (المخرجات)
class ProcessedSalesSchema(pa.DataFrameModel):
    """مخطط البيانات بعد المعالجة."""
    order_id: Series[int] = pa.Field(nullable=False, unique=True)
    product_name: Series[str] = pa.Field(nullable=False)
    price: Series[float] = pa.Field(gt=0, nullable=False)
    quantity: Series[int] = pa.Field(ge=1, nullable=False)
    total: Series[float] = pa.Field(gt=0, nullable=False)

    class Config:
        strict = False
        coerce = True

# التحقق من المدخلات فقط
@check_input(RawSalesSchema)
def load_sales_data(df: pd.DataFrame) -> pd.DataFrame:
    """تحميل البيانات مع التحقق من صحة المدخلات."""
    print(f"تم تحميل {len(df)} صف من البيانات")
    return df

# التحقق من المخرجات فقط
@check_output(ProcessedSalesSchema)
def process_sales(df: pd.DataFrame) -> pd.DataFrame:
    """معالجة بيانات المبيعات مع التحقق من صحة المخرجات."""
    df = df.drop_duplicates(subset=["order_id"])
    df = df[df["price"] > 0].copy()
    df["total"] = df["price"] * df["quantity"]
    return df

# التحقق من المدخلات والمخرجات معًا
@check_io(
    df=check_input(RawSalesSchema),
    out=check_output(ProcessedSalesSchema)
)
def full_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    """أنبوب معالجة كامل مع التحقق من المدخلات والمخرجات."""
    df = df.drop_duplicates(subset=["order_id"])
    df = df[df["price"] > 0].copy()
    df["total"] = df["price"] * df["quantity"]
    return df

# تشغيل أنبوب المعالجة
raw_data = pd.DataFrame({
    "order_id": [1001, 1002, 1003, 1004],
    "product_name": ["حاسوب محمول", "هاتف ذكي", "سماعات", "شاشة"],
    "price": [2500.0, 1200.0, 150.0, 3500.0],
    "quantity": [1, 2, 5, 1],
})

# يتم التحقق تلقائيًا عند استدعاء الدالة
result = full_pipeline(raw_data)
print("تمت المعالجة بنجاح!")
print(result)

هالنمط قوي جدًا لأنه يحول التحقق من صحة البيانات إلى جزء لا يتجزأ من منطق المعالجة. ما تحتاج تتذكر تستدعي validate() يدويًا — المُزخرفات تتكفل بكل شي تلقائيًا.

مثال عملي: التحقق من بيانات المبيعات في أنبوب ETL

خلنا الحين نبني مثال عملي شامل يحاكي سيناريو واقعي: أنبوب ETL (استخراج، تحويل، تحميل) لمعالجة بيانات مبيعات متجر إلكتروني. الفكرة هي إننا نتحقق من صحة البيانات في كل مرحلة — عند الاستيعاب، وبعد التحويل، وقبل التصدير.

import pandas as pd
import pandera as pa
from pandera.typing import Series
from pandera import check_input, check_output
import numpy as np
from datetime import datetime
import logging

# إعداد نظام التسجيل
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("etl_pipeline")

# ============================================================
# المرحلة 1: تعريف المخططات لكل مرحلة من مراحل الأنبوب
# ============================================================

class RawDataSchema(pa.DataFrameModel):
    """مخطط التحقق من البيانات الخام عند الاستيعاب."""
    order_id: Series[int] = pa.Field(nullable=False)
    product: Series[str] = pa.Field(nullable=False, str_length={"min_value": 1})
    price: Series[float] = pa.Field(nullable=False)
    qty: Series[int] = pa.Field(nullable=False)
    date: Series[str] = pa.Field(nullable=False)
    region: Series[str] = pa.Field(nullable=False)

    class Config:
        strict = False
        coerce = True
        name = "مخطط البيانات الخام"

class CleanDataSchema(pa.DataFrameModel):
    """مخطط التحقق من البيانات بعد التنظيف."""
    order_id: Series[int] = pa.Field(nullable=False, unique=True, ge=1)
    product: Series[str] = pa.Field(nullable=False, str_length={"min_value": 1})
    price: Series[float] = pa.Field(gt=0, lt=500_000, nullable=False)
    qty: Series[int] = pa.Field(ge=1, le=1000, nullable=False)
    date: Series[pa.DateTime] = pa.Field(nullable=False)
    region: Series[str] = pa.Field(
        nullable=False,
        isin=["الرياض", "جدة", "الدمام", "مكة", "المدينة"]
    )
    total: Series[float] = pa.Field(gt=0, nullable=False)

    class Config:
        strict = False
        coerce = True
        name = "مخطط البيانات النظيفة"

    # فحص مخصص: التأكد من أن المجموع = السعر × الكمية
    @pa.dataframe_check
    def total_equals_price_times_qty(cls, df: pd.DataFrame) -> bool:
        """التحقق من صحة حساب المجموع."""
        expected = (df["price"] * df["qty"]).round(2)
        actual = df["total"].round(2)
        return (expected == actual).all()

# ============================================================
# المرحلة 2: دوال المعالجة مع التحقق التلقائي
# ============================================================

@check_input(RawDataSchema)
def extract_data(df: pd.DataFrame) -> pd.DataFrame:
    """استخراج البيانات الخام والتحقق من بنيتها الأساسية."""
    logger.info(f"استخراج البيانات: {len(df)} صف")
    return df

def transform_data(df: pd.DataFrame) -> pd.DataFrame:
    """تنظيف وتحويل البيانات."""
    logger.info("بدء عملية التحويل...")
    initial_count = len(df)

    # إزالة التكرارات
    df = df.drop_duplicates(subset=["order_id"])
    logger.info(f"بعد إزالة التكرارات: {len(df)} صف (أُزيل {initial_count - len(df)})")

    # إزالة الصفوف ذات الأسعار غير المنطقية
    df = df[df["price"] > 0].copy()

    # إزالة الصفوف ذات الكميات غير المنطقية
    df = df[df["qty"] >= 1].copy()

    # تنظيف أسماء المنتجات
    df["product"] = df["product"].str.strip().str.lower()

    # تنظيف أسماء المناطق
    df["region"] = df["region"].str.strip()

    # تحويل التاريخ
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df = df.dropna(subset=["date"])

    # حساب المجموع
    df["total"] = (df["price"] * df["qty"]).round(2)

    logger.info(f"بعد التحويل: {len(df)} صف")
    return df

@check_output(CleanDataSchema, lazy=True)
def validate_and_load(df: pd.DataFrame) -> pd.DataFrame:
    """التحقق النهائي من البيانات المعالجة قبل التحميل."""
    logger.info("التحقق النهائي من البيانات...")
    return df

# ============================================================
# المرحلة 3: تشغيل أنبوب ETL الكامل
# ============================================================

def run_etl_pipeline():
    """تشغيل أنبوب ETL الكامل مع التحقق في كل مرحلة."""

    # محاكاة بيانات خام واقعية تحتوي على مشاكل
    np.random.seed(42)
    raw_df = pd.DataFrame({
        "order_id": [101, 102, 103, 104, 105, 106, 107, 108, 108, 110],
        "product": [
            "حاسوب محمول", "  هاتف ذكي  ", "سماعات", "شاشة",
            "لوحة مفاتيح", "فأرة", "طابعة", "كاميرا", "كاميرا", "حاسوب لوحي"
        ],
        "price": [4500.0, 2800.0, 350.0, 6200.0, 180.0, -50.0, 1200.0, 3500.0, 3500.0, 2200.0],
        "qty": [1, 2, 5, 1, 3, 1, 1, 2, 2, 0],
        "date": [
            "2026-01-15", "2026-01-16", "2026-01-17", "2026-01-18",
            "2026-01-19", "2026-01-20", "2026-01-21", "2026-01-22",
            "2026-01-22", "2026-01-23"
        ],
        "region": [
            "الرياض", "جدة", "الدمام", "مكة",
            "المدينة", "الرياض", "جدة ", "الدمام",
            "الدمام", "الرياض"
        ],
    })

    logger.info("=" * 60)
    logger.info("بدء أنبوب ETL")
    logger.info("=" * 60)

    try:
        # الاستخراج مع التحقق من البنية
        extracted = extract_data(raw_df)

        # التحويل والتنظيف
        transformed = transform_data(extracted)

        # التحقق النهائي والتحميل
        final_data = validate_and_load(transformed)

        # حفظ النتيجة
        final_data.to_parquet("sales_validated.parquet", index=False)

        logger.info("=" * 60)
        logger.info(f"تم بنجاح! {len(final_data)} صف من البيانات النظيفة والموثقة")
        logger.info(f"إجمالي المبيعات: {final_data['total'].sum():,.2f} ريال")
        logger.info("=" * 60)

        return final_data

    except pa.errors.SchemaErrors as exc:
        logger.error("فشل التحقق من صحة البيانات!")
        logger.error(f"عدد الأخطاء: {len(exc.failure_cases)}")
        logger.error(f"تفاصيل الأخطاء:\n{exc.failure_cases}")
        raise
    except pa.errors.SchemaError as exc:
        logger.error(f"خطأ في المخطط: {exc}")
        raise

# تشغيل الأنبوب
result = run_etl_pipeline()
print("\nالبيانات النهائية:")
print(result.to_string(index=False))

هالمثال يوضح كيف تشتغل Pandera كطبقة حماية في كل مرحلة من مراحل أنبوب المعالجة. عند الاستيعاب: هل البيانات الخام لها البنية المتوقعة؟ بعد التحويل: هل التنظيف أنتج بيانات صالحة؟ قبل التحميل: هل البيانات النهائية مطابقة لعقد البيانات؟

إذا حصل أي انحراف في أي مرحلة، الأنبوب يتوقف فورًا مع رسالة خطأ واضحة بدل ما يُنتج نتائج خاطئة بصمت. وهذا بالضبط اللي تبيه في بيئة إنتاج حقيقية.

استخدام Pandera في بيئة الإنتاج

Pandera مو بس أداة للتطوير — هي مصممة للعمل في بيئات الإنتاج الفعلية. خلني أشاركك أبرز أنماط الاستخدام:

التكامل مع CI/CD

تقدر تضيف اختبارات التحقق من البيانات كجزء من خط أنابيب التكامل والنشر المستمر. إذا تغيرت بنية البيانات أو انتهكت القواعد، الاختبار يفشل قبل ما يوصل لبيئة الإنتاج. وهالشي بيوفر عليك ساعات من تنقيح الأخطاء.

# test_data_validation.py — اختبار يمكن تشغيله في CI/CD
import pytest
import pandas as pd
import pandera as pa
from pandera.typing import Series

class ExpectedSchema(pa.DataFrameModel):
    user_id: Series[int] = pa.Field(unique=True, ge=1)
    score: Series[float] = pa.Field(ge=0, le=100)
    status: Series[str] = pa.Field(isin=["active", "inactive"])

    class Config:
        coerce = True

def test_production_data_matches_schema():
    """اختبار أن بيانات الإنتاج تطابق المخطط المتوقع."""
    df = pd.read_csv("data/production_sample.csv")
    ExpectedSchema.validate(df, lazy=True)

def test_schema_handles_empty_dataframe():
    """اختبار أن المخطط يتعامل بشكل صحيح مع DataFrame فارغ."""
    empty_df = pd.DataFrame(columns=["user_id", "score", "status"])
    result = ExpectedSchema.validate(empty_df)
    assert len(result) == 0

def test_schema_rejects_invalid_data():
    """اختبار أن المخطط يرفض البيانات غير الصالحة."""
    invalid_df = pd.DataFrame({
        "user_id": [1, 1],       # معرّف مكرر
        "score": [50.0, -10.0],  # درجة سالبة
        "status": ["active", "unknown"],  # حالة غير معروفة
    })
    with pytest.raises(pa.errors.SchemaErrors):
        ExpectedSchema.validate(invalid_df, lazy=True)

كشف انحراف البيانات (Data Drift Detection)

من الاستخدامات الذكية لـ Pandera: كشف التغيرات غير المتوقعة في توزيع البيانات بمرور الوقت. تقدر تستخدم الفحوصات الإحصائية لهالغرض:

import pandera as pa
from pandera.typing import Series

class MonitoredSchema(pa.DataFrameModel):
    """مخطط مراقبة يكشف انحراف البيانات."""
    price: Series[float] = pa.Field(gt=0)
    quantity: Series[int] = pa.Field(ge=1)

    # كشف انحراف المتوسط
    @pa.check("price")
    def price_mean_stable(cls, series):
        """التحقق من أن متوسط السعر ضمن النطاق التاريخي."""
        return 500 <= series.mean() <= 5000

    # كشف انحراف التوزيع
    @pa.check("quantity")
    def quantity_distribution_stable(cls, series):
        """التحقق من أن توزيع الكمية لم ينحرف بشكل كبير."""
        return series.std() < series.mean() * 2  # الانحراف المعياري لا يتجاوز ضعف المتوسط

Pandera مقابل Pydantic — متى تستخدم كل منهما؟

هالسؤال يجيني كثير: ليش ما نستخدم Pydantic بدل Pandera؟ الجواب المختصر هو إنهم أداتين لمهمتين مختلفتين، ويمكنهم يشتغلون مع بعض بشكل ممتاز.

Pandera مصممة خصيصًا للتحقق من صحة إطارات البيانات (DataFrames). تفهم بنية الأعمدة والصفوف والسلاسل، وتوفر فحوصات إحصائية على مستوى العمود بالكامل. هي الخيار الأمثل للبيانات الجدولية الضخمة.
Pydantic مصممة للتحقق من صحة الكائنات (Objects) وبيانات JSON. ممتازة للتحقق من إعدادات التطبيق، واستجابات واجهات API، وبيانات الطلبات في تطبيقات الويب.

from pydantic import BaseModel, field_validator
import pandera as pa
from pandera.typing import Series, DataFrame
import pandas as pd

# Pydantic: للتحقق من إعدادات أنبوب المعالجة
class PipelineConfig(BaseModel):
    """إعدادات أنبوب المعالجة — التحقق بواسطة Pydantic."""
    input_path: str
    output_path: str
    min_price: float = 0.0
    max_rows: int = 1_000_000

    @field_validator("min_price")
    @classmethod
    def price_must_be_positive(cls, v):
        if v < 0:
            raise ValueError("الحد الأدنى للسعر يجب أن يكون غير سالب")
        return v

# Pandera: للتحقق من البيانات الجدولية
class SalesDataSchema(pa.DataFrameModel):
    """مخطط بيانات المبيعات — التحقق بواسطة Pandera."""
    order_id: Series[int] = pa.Field(unique=True)
    price: Series[float] = pa.Field(gt=0)
    quantity: Series[int] = pa.Field(ge=1)

    class Config:
        coerce = True

# الاستخدام المشترك: Pydantic للإعدادات + Pandera للبيانات
def run_validated_pipeline(config_dict: dict, data: pd.DataFrame):
    """أنبوب يستخدم كلتا المكتبتين معًا."""
    # التحقق من الإعدادات بـ Pydantic
    config = PipelineConfig(**config_dict)

    # التحقق من البيانات بـ Pandera
    validated_data = SalesDataSchema.validate(data)

    # تطبيق الإعدادات على البيانات الموثقة
    filtered = validated_data[validated_data["price"] >= config.min_price]
    filtered = filtered.head(config.max_rows)

    filtered.to_parquet(config.output_path, index=False)
    return filtered

الخلاصة: استخدم Pydantic للإعدادات واستجابات API والكائنات المفردة، واستخدم Pandera لإطارات البيانات والبيانات الجدولية. في المشاريع الحقيقية، غالبًا بتلاقي نفسك تستخدم الاثنين مع بعض — وهذا شي طبيعي تمامًا.

الأسئلة الشائعة

هل يمكن استخدام Pandera مع مكتبات غير pandas؟

أكيد نعم. اعتبارًا من الإصدار 0.29.0، Pandera تدعم ستة محركات معالجة بيانات: pandas وPolars وDask وModin وIbis وPySpark. التصميم الأساسي للمخططات يبقى متشابه عبر جميع المحركات، وهذا يعني إنك تقدر تنقل منطق التحقق بين المحركات مع تعديلات طفيفة فقط. لتثبيت الدعم لمحرك معين، استخدم pip install 'pandera[polars]' أو المحرك المطلوب.

ما الفرق بين DataFrameSchema و DataFrameModel؟

كلاهما يحقق نفس الغرض، لكن بأسلوبين مختلفين. DataFrameSchema يستخدم أسلوب القاموس وهو مناسب لما تحتاج تبني المخطط ديناميكيًا في وقت التشغيل. DataFrameModel يستخدم أسلوب الفئات (مثل Pydantic) وهو المفضل لأنه أوضح في القراءة، يدعم الوراثة، ويتكامل أفضل مع أدوات تحليل الأنواع الثابتة.

المشاريع الجديدة يُنصح بالبدء مع DataFrameModel مباشرة.

كيف أتعامل مع الأداء عند التحقق من ملايين الصفوف؟

Pandera مصممة لتكون سريعة، لكن التحقق من مجموعات بيانات ضخمة ممكن يستغرق وقت ملحوظ. إليك بعض النصائح العملية: أولًا، استخدم coerce=True بحذر لأن تحويل الأنواع يستهلك موارد إضافية. ثانيًا، تحقق من عينة من البيانات (df.sample(n=10000)) في بيئة التطوير بدل التحقق الكامل. ثالثًا، في بيئة الإنتاج، فعّل التحقق السريع (بدون lazy=True) عشان يتوقف عند أول خطأ. رابعًا، فكر في استخدام Pandera مع Dask أو PySpark لتوزيع عملية التحقق على عدة عقد.

هل يمكنني استخدام Pandera لاختبار الفرضيات الإحصائية؟

نعم، Pandera توفر دعم مدمج لاختبار الفرضيات الإحصائية من خلال pa.Hypothesis. تقدر تختبر إذا كان متوسط عمودين مختلف إحصائيًا (اختبار t)، أو إذا كان عمود يتبع توزيع معين. هالشي مفيد بشكل خاص لكشف انحراف البيانات (data drift) في بيئة الإنتاج.

هل Pandera بديل عن Great Expectations؟

كلاهما يشتغل في مجال التحقق من صحة البيانات، لكنهم يناسبون مقاييس مختلفة. Pandera مكتبة خفيفة تتكامل مباشرة في كود Python، وهي مثالية لأنابيب المعالجة في المشاريع الصغيرة والمتوسطة ومشاريع التعلم الآلي. Great Expectations منصة مؤسسية كاملة مع واجهة ويب وتكامل مع منسقي المهام مثل Airflow وDagster.

إذا كنت تشتغل على مشروع Python أو خدمة صغيرة، Pandera هي الخيار الأبسط والأكثر كفاءة. أما للفرق الكبيرة اللي تدير عشرات الأنابيب، فـ Great Expectations ممكن يكون أنسب. شخصيًا، أبدأ دايمًا بـ Pandera وأنتقل لـ Great Expectations بس إذا المشروع كبر فعلًا.