قالب Dataflow من Pub/Sub إلى ClickHouse

قالب Pub/Sub إلى ClickHouse هو مسار معالجة متدفق يقرأ الرسائل المُرمَّزة بصيغة JSON من اشتراك في Pub/Sub ويكتبها في جدول ClickHouse. تُوجَّه الرسائل التي يتعذر parse لها أو يتعذر map لها إلى schema الهدف إلى وجهة الرسائل المتعذّرة: جدول ClickHouse أو topic في Pub/Sub أو كليهما.

• يجب أن يكون اشتراك Pub/Sub المصدر موجودًا.
• يجب أن تكون الرسائل المنشورة إلى الاشتراك بتنسيق JSON صالح.
• يجب أن يكون جدول ClickHouse المستهدف موجودًا، وأن تتطابق أسماء أعمدته مع أسماء الحقول في حمولة JSON.
• يجب أن يكون مضيف ClickHouse متاحًا من أجهزة العامل في Dataflow.
• يجب توفير وجهة واحدة على الأقل للرسائل المتعذّرة (clickHouseDeadLetterTable أو deadLetterTopic). وإذا تم توفير كلتيهما، فسيتم توجيه الرسائل الفاشلة إلى الوجهتين كلتيهما في الوقت نفسه.
• عند تعيين clickHouseDeadLetterTable، يجب أن يكون جدول الرسائل المتعذّرة موجودًا مسبقًا في ClickHouse بالمخطط الموضّح في معالجة الرسائل المتعذّرة.
• عند تعيين deadLetterTopic، يجب أن يكون موضوع Pub/Sub موجودًا مسبقًا.

يجب أن تكون رسائل Pub/Sub كائنات JSON تتطابق أسماء حقولها ذات المستوى الأعلى تمامًا مع أسماء أعمدة جدول ClickHouse الهدف.

لمواءمة الرسائل الواردة مع الجدول الهدف، ينفّذ مسار المعالجة ما يلي عند بدء التشغيل:

1. يجلب مخطط جدول ClickHouse الهدف.
2. يبني مخطط Row في Beam انطلاقًا من مخطط ClickHouse هذا.
3. لكل رسالة Pub/Sub واردة، يحلّل حمولة JSON ويُنشئ صفًا بقراءة الحقول المسمّاة في مخطط ClickHouse.

تُحوَّل قيم JSON إلى نوع عمود ClickHouse المقابل:

نظرًا لأن مسار المعالجة تدفقي، تتراكم الصفوف الواردة ضمن نوافذ قبل كتابتها إلى ClickHouse. وتُختار استراتيجية تحديد النوافذ بناءً على المعلمات التي توفّرها:

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

تُوجَّه الرسائل التي تفشل في تحليل JSON، أو تعيين المخطط، أو تحويل النوع قسرًا إلى وجهة أو وجهات الرسائل المتعذّرة المُعدّة. ويجب توفير واحد على الأقل من clickHouseDeadLetterTable أو deadLetterTopic؛ وإذا جرى تعيينهما معًا، فستُرسَل الرسائل الفاشلة إلى كليهما.

عند تعيين clickHouseDeadLetterTable، يجب أن يكون جدول الرسائل المتعذّرة موجودًا مسبقًا بهذا المخطط الثابت:

تعريف مبسّط لعملية نشر بعقدة واحدة:

CREATE TABLE my_table_dead_letter (
    raw_message   String,
    error_message String,
    stack_trace   String,
    failed_at     DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;

عند تعيين deadLetterTopic، تُعاد كل رسالة فاشلة إلى الموضوع مع:

• الحمولة: بايتات الرسالة الأصلية.
• السمة errorMessage: رسالة الاستثناء التي تم التقاطها وقت حدوث الفشل.
• السمة failedAt: الطابع الزمني لوقت المعالجة الذي فشل عنده الصف.

وهذا يسهّل إعادة تمرير الرسائل الفاشلة بعد حل المشكلة الأساسية في المخطط أو المُنتِج.

يتوفر قالب Pub/Sub إلى ClickHouse عبر Google Cloud Console.

سجّل الدخول إلى Google Cloud Console وابحث عن Dataflow.

1. اضغط على الزر CREATE JOB FROM TEMPLATE.

   

2. بعد فتح نموذج القالب، أدخل اسم المهمة وحدد المنطقة المطلوبة.

3. في حقل Dataflow Template، اكتب ClickHouse أو Pub/Sub، ثم اختر قالب Pub/Sub to ClickHouse.

4. بعد تحديده، يتوسع النموذج. املأ ما يلي:

   • اشتراك إدخال Pub/Sub، بالصيغة projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
   • عنوان URL لنقطة نهاية ClickHouse — ولاستخدام ClickHouse Cloud، استخدم https://<HOST>:8443.
   • قاعدة بيانات ClickHouse، والجدول الهدف، واسم المستخدم وكلمة المرور.
   • وجهة الرسائل المتعذّرة واحدة على الأقل: جدول ClickHouse أو موضوع في Pub/Sub (أو كلاهما).

5. يمكنك اختياريًا تخصيص التجميع على دفعات (windowSeconds, batchRowCount) ومعلمات ضبط ClickHouseIO، كما هو موضح في قسم Template parameters.

انتقل إلى علامة التبويب Dataflow Jobs في Google Cloud Console لمتابعة حالة المهمة. ستجد تفاصيل المهمة، بما في ذلك مدى التقدّم وأي أخطاء:

يُصدر القالب أيضًا المقاييس المخصّصة التالية ضمن مساحة الاسم PubSubToClickHouse، ويمكن الاطلاع عليها من صفحة مهمة Dataflow:

يحدث هذا الخطأ عندما تنفد ذاكرة ClickHouse أثناء معالجة دفعات كبيرة من البيانات. لحل هذه المشكلة:

• زيادة موارد المثيل: قم بترقية ClickHouse server إلى مثيل أكبر بذاكرة أكثر للتعامل مع حمل معالجة البيانات.
• تقليل حجم الدفعة: خفّض batchRowCount (و/أو maxInsertBlockSize) في إعدادات مهمة Dataflow لإرسال أجزاء أصغر من البيانات إلى ClickHouse، مما يقلّل استهلاك الذاكرة لكل دفعة.

الأسباب الأكثر شيوعًا هي:

• أسماء حقول JSON لا تتطابق تمامًا مع أسماء أعمدة ClickHouse (والتطابق هنا حساس لحالة الأحرف).
• يتعذر تحويل قيمة JSON إلى نوع العمود المطلوب (على سبيل المثال، سلسلة غير متوافقة مع ISO-8601 في عمود DateTime).
• تغيّر مخطط الجدول الهدف منذ بدء مسار المعالجة — إذ يُجلَب المخطط مرة واحدة فقط عند بدء التشغيل. أعد تشغيل المهمة بعد تطبيق تغييرات المخطط.

افحص العمودين error_message وstack_trace في جدول ClickHouse الخاص بالرسائل المتعذّرة (أو السمة errorMessage في رسائل Pub/Sub المتعذّرة) لتحديد السبب الجذري.

• تأكَّد من أن الاشتراك يستقبل الرسائل — تحقّق من مقياس messages-received في صفحة مهمة Dataflow.
• في الوضع المعتمد على الوقت (windowSeconds فقط)، لا تُفرَّغ الصفوف إلا عند حدود النافذة. خفِّض قيمة windowSeconds للتحقّق من حدوث عمليات التفريغ.
• تحقّق من إمكانية الوصول الشبكي بين عمّال Dataflow ونقطة نهاية ClickHouse (firewall أو VPC peering أو Private Service Connect).

الكود المصدري للقالب متاح في:

• GoogleCloudPlatform/DataflowTemplates — مستودع Google Cloud Platform الأصلي.
• ClickHouse/DataflowTemplates — النسخة المتفرعة الخاصة بـ ClickHouse.