انتقل إلى المحتوى الرئيسي

الوحدة 7 — Feast عمليًّا

كلّ ما سبق مبدأ. هذه الوحدة تُنفّذه بـFeast على مشروع الاحتيال، بأصغر تركيبة تعمل من طرف إلى طرف: Parquet محلّيًّا للمخزن غير المتّصل، Redis محلّيًّا للمخزن المتّصل، وسجلّ ملفّ. الأدوات نفسها تعمل بنفس الأوامر حين تنتقل إلى BigQuery وDynamoDB في الإنتاج.

بنية دفتر المتغيّرات​

feast init fraud_repo يُنشئ الهيكل الآتي:

fraud_repo/
├── feature_store.yaml
├── data/
│   └── transactions.parquet
└── features/
    └── transactions.py

feature_store.yaml يُصرّح الاستضافة:

project: fraud_scoring
registry: data/registry.db
provider: local
online_store:
  type: redis
  connection_string: "localhost:6379"
offline_store:
  type: file
entity_key_serialization_version: 2

الانتقال إلى الإنتاج: يُغيَّر provider إلى aws أو gcp، ويُعدَّل online_store إلى DynamoDB، ويُغيَّر offline_store إلى BigQuery. بقيّة الشفرة تبقى كما هي.

تعريف الكيانات والوصفات​

في features/transactions.py:

from datetime import timedelta
from feast import Entity, Field, FeatureView, FileSource
from feast.types import Int64, Float32

customer = Entity(name="customer", join_keys=["customer_id"])

transactions = FileSource(
    name="transactions",
    path="data/transactions.parquet",
    timestamp_field="event_time",
    created_timestamp_column="created_at",
)

customer_counters = FeatureView(
    name="customer_counters",
    entities=[customer],
    ttl=timedelta(hours=2),
    schema=[
        Field(name="count_1h", dtype=Int64),
        Field(name="count_24h", dtype=Int64),
        Field(name="count_7d", dtype=Int64),
        Field(name="avg_amount_7d", dtype=Float32),
    ],
    source=transactions,
    online=True,
    tags={"owner": "fraud-team", "domain": "payments"},
)

created_timestamp_column مقصود: يميّز وقت الحدث (متى وقعت المعاملة) عن وقت الوصول (متى وصلت إلى النظام). Feast يستخدم هذا في الدمج الزمنيّ لتفادي تسرّب حين يتأخّر وصول حدث.

تسجيل التعاريف: feast apply​

cd fraud_repo
feast apply

هذا الأمر يقرأ ملفّ Python، يُصادِق التعريفات، ويكتبها في السجلّ (registry.db). كلّ تعديل لوصفة يمرّ عبر هذا الأمر. في CI/CD، يُشغَّل feast apply بعد كلّ دمج إلى الفرع الرئيس، على مستودع مركزيّ يعرفه كلّ المستهلكين.

بعد apply، السجلّ يعرف الكيانات والوصفات، لكنّ أيّ مخزن (متّصل أو لا) لم يتغيّر بعد.

استخراج مجموعة تدريب: get_historical_features​

from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="fraud_repo")

entity_df = pd.read_parquet("data/labels.parquet")[
    ["customer_id", "event_time", "is_fraud"]
]

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "customer_counters:count_1h",
        "customer_counters:count_24h",
        "customer_counters:count_7d",
        "customer_counters:avg_amount_7d",
    ],
).to_df()

خلف الكواليس، Feast يُشغّل دمجًا زمنيًّا صحيحًا (الوحدة 5) على المخزن غير المتّصل. النتيجة training_df جاهزة لـxgboost أو أيّ نموذج آخر، بلا تسرّب مستقبل.

قبل التصدير إلى الإنتاج، هذه الخطوة الوحيدة لمُهندسة تعلّم آليّ. لم تعد تكتب SQL دمج، ولم تعد تتذكّر منطقة زمنيّة، ولم يعد تعريف العدّاد ملكها.

ملء المخزن المتّصل: materialize-incremental​

feast materialize-incremental $(date -Iseconds)

يُشغَّل هذا مرّة أولى بامتداد كبير (مثلًا سبعة أيّام)، ثمّ يُوضع في cron كلّ 5 دقائق. Feast يقرأ Parquet، يحسب الوصفة عن الفترة الجديدة، ويكتب النتائج في Redis.

بعد التنفيذ، Redis يحتوي مفاتيح من شكل fraud_scoring:customer_counters:{customer_id} تحمل حزمة القيم الأربع.

خدمة القيم عند الاستدلال: get_online_features​

في مُخدَم الاستدلال (Flask أو FastAPI):

from feast import FeatureStore

store = FeatureStore(repo_path="fraud_repo")

def predict(request):
    features = store.get_online_features(
        features=[
            "customer_counters:count_1h",
            "customer_counters:count_24h",
            "customer_counters:count_7d",
            "customer_counters:avg_amount_7d",
        ],
        entity_rows=[{"customer_id": request["customer_id"]}],
    ).to_dict()

    x = [
        features["count_1h"][0],
        features["count_24h"][0],
        features["count_7d"][0],
        features["avg_amount_7d"][0],
    ]
    return model.predict_proba([x])[0][1]

الاستدعاء الأخير على Redis محلّيًّا يستغرق حوالي 1 مليّ‌ثانية. الأهمّ: أسماء الحقول (count_1h…) هي نفسها كما في التدريب، لأنّها تأتي من نفس الوصفة customer_counters. لا مجال لخطأ ترميز أو منطقة زمنيّة.

خادم مستقلّ: feast serve​

بديل: تشغيل خادم متغيّرات مستقلّ ونداؤه عبر REST:

feast serve --host 0.0.0.0 --port 6566

ثمّ من أيّ مُخدَم استدلال:

curl -X POST http://localhost:6566/get-online-features \
  -H "Content-Type: application/json" \
  -d '{
    "features": ["customer_counters:count_1h"],
    "entities": {"customer_id": [42]}
  }'

هذا يُنظّف مُخدَم الاستدلال من أيّ اعتماد مباشر على Redis أو Feast SDK. المُخدَم يحتاج فقط عنوانًا وواجهة HTTP.

حدود Feast​

Feast ليس كلّ شيء. عليك أن تعرف ما لا يفعله:

  • لا يحسب تدفّقات بنفسه. تحتاج Spark Streaming أو Flink يكتب في push_source.
  • لا يوفّر لوحة اكتشاف كاملة كـTecton. مبادرات Feast UI تتطوّر لكن تبقى بسيطة.
  • إدارة الأذونات دقيقة: تُدار على مستوى التخزين السفليّ (IAM لـS3، ACL لـRedis)، لا داخل Feast.
  • لا مراقبة انحراف مدمجة: تحتاج أدوات إضافيّة (الوحدة 9).

في المقابل، هو مفتوح المصدر، ويعمل بلا اعتماد على بائع، ويسمح بالانتقال من مختبر محلّيّ إلى إنتاج ضخم دون إعادة كتابة الشفرة.

اختبار قبل النشر

بعد كلّ feast apply جديد، شغّل اختبار «الرحلة الكاملة»: get_historical_features على عيّنة، materialize-incremental على فترة قصيرة، ثمّ get_online_features مباشرة. مقارنة النتائج على كيان محدّد تكشف أيّ فجوة أدخلها التغيير.

الخلاصة​

  • Feast يعتمد أربعة أوامر: apply (السجلّ)، get_historical_features (التدريب)، materialize-incremental (المخزن المتّصل)، get_online_features (الخدمة).
  • التصميم يعزل التعاريف عن اختيار الاستضافة، فيمكن التنقّل من محلّيّ إلى سحابيّ من دون تغيير الشفرة.
  • المتغيّرات عند الاستدلال تحمل نفس الأسماء والأنواع كما في التدريب، فيسقط أهمّ سبب للفجوة.
  • Feast يوفّر البنية لا كلّ الحلقة: التدفّق، والمراقبة، والحوكمة تُبنى فوقه.

الوحدة التالية: كيف يُعاد استعمال متغيّرات فريق الاحتيال في فريق التسويق بلا ازدواجيّة ولا فوضى.