دعونا نستخدم جدول حقائق الاشتراك كمثال هنا. يتم تشغيل سير عمل هذا الجدول كل ساعة، حيث يكون مصدر الإدخال الرئيسي هو جدول Iceberg الذي يخزن جميع أحداث التسجيل الأولية مقسمة حسب تاريخ الوصول والساعة ومعرف الدُفعة.
فيما يلي مقتطف YAML يوضح التكوين لهذا أثناء خطوة تهيئة Psyberg:
- job:
id: psyberg_session_init
type: Spark
spark:
app_args:
- --process_name=signup_fact_load
- --src_tables=raw_signups
- --psyberg_session_id=20230914061001
- --psyberg_hwm_table=high_water_mark_table
- --psyberg_session_table=psyberg_session_metadata
- --etl_pattern_id=1
خلف الكواليس، يحدد Psyberg أن خط الأنابيب هذا تم تكوينه لنمط عديم الحالة منذ ذلك الحين etl_pattern_id=1.
يستخدم Psyberg أيضًا المدخلات المتوفرة لاكتشاف لقطات Iceberg التي استمرت بعد أحدث علامة مائية عالية متوفرة في جدول العلامة المائية. باستخدام عمود الملخص في بيانات تعريف اللقطة [see the Iceberg Metadata section in post 1 for more details]، نقوم بتحليل معلومات القسم لكل لقطة Iceberg من الجدول المصدر.
بعد ذلك، يحتفظ Psyberg بمعرفات URI الخاصة بالمعالجة (مجموعة من سلاسل JSON تحتوي على مجموعات من معرفات تاريخ الوصول والساعة ومعرفات الدُفعات) كما هو محدد بواسطة تغييرات اللقطة. يتم تخزين هذه المعلومات وبيانات التعريف المحسوبة الأخرى في ملف psyberg_session_f طاولة. هذه البيانات المخزنة متاحة بعد ذلك لللاحقة LOAD.FACT_TABLE وظيفة في سير العمل للاستفادة ولأغراض التحليل والتصحيح.
يتم استخدام معالجة البيانات ذات الحالة عندما يعتمد الإخراج على سلسلة من الأحداث عبر تدفق إدخال واحد أو أكثر.
لنأخذ بعين الاعتبار مثال إنشاء جدول حقائق الإلغاء، والذي يأخذ ما يلي كمدخلات:
- أحداث الإلغاء الخام تشير إلى متى تم إلغاء حساب العميل
- جدول الحقيقة الذي يخزن الوارد طلبات العملاء لإلغاء اشتراكهم في نهاية فترة الفاتورة
تساعد هذه المدخلات في استخلاص سمات تحليلية إضافية مثل نوع التغيير، أي طوعي أو غير طوعي، وما إلى ذلك.
تختلف خطوة التهيئة لمعالجة البيانات ذات الحالة قليلاً عن عديمة الحالة. تقدم Psyberg تكوينات إضافية وفقًا لاحتياجات خطوط الأنابيب. فيما يلي مقتطف YAML يوضح تكوين جدول حقائق الإلغاء أثناء خطوة تهيئة Psyberg:
- job:
id: psyberg_session_init
type: Spark
spark:
app_args:
- --process_name=cancel_fact_load
- --src_tables=raw_cancels|processing_ts,cancel_request_fact
- --psyberg_session_id=20230914061501
- --psyberg_hwm_table=high_water_mark_table
- --psyberg_session_table=psyberg_session_metadata
- --etl_pattern_id=2
خلف الكواليس، يحدد Psyberg أن خط الأنابيب هذا تم تكوينه لنمط ذو حالة منذ ذلك الحين etl_pattern_id هو 2.
لاحظ التفاصيل الإضافية في قائمة src_tables المقابلة لـ Raw_cancels أعلاه. ال معالجة_ts يمثل هنا الطابع الزمني لمعالجة الحدث والذي يختلف عن الطابع الزمني المعتاد لالتزام لقطة Iceberg، أي events_landing_ts كما هو موضح في الجزء 1 من هذه السلسلة.
ومن المهم التقاط نطاق مجموعة الأحداث الموحدة من جميع المصادر، أي كل من الإلغاءات الأولية وإلغاء_طلب_الأحداث، مع الأخذ في الاعتبار الأحداث التي وصلت متأخرة. يمكن تتبع التغييرات التي يتم إجراؤها على لقطات الجدول المصدر باستخدام حقول الطابع الزمني المختلفة. معرفة حقل الطابع الزمني الذي سيتم استخدامه على سبيل المثال events_landing_ts أو شيء من هذا القبيل معالجة_ts يساعد على تجنب الأحداث المفقودة.
على غرار النهج المتبع في معالجة البيانات عديمة الحالة، يستخدم Psyberg المدخلات المتوفرة لتحليل معلومات القسم لكل لقطة Iceberg في الجدول المصدر.
