CDC من DynamoDB إلى ClickHouse

يتيح لك ClickPipes توصيل ClickHouse بـ DynamoDB.

تشرح هذه الصفحة كيفية إعداد CDC من DynamoDB إلى ClickHouse باستخدام ClickPipes. ويتكوّن هذا التكامل من عنصرين:

1. اللقطة الأولية عبر S3 ClickPipes
2. التحديثات في الوقت الفعلي عبر Kinesis ClickPipes

سيتم استيعاب البيانات في ReplacingMergeTree. ويُستخدم table engine هذا عادةً في سيناريوهات CDC للسماح بتطبيق عمليات التحديث. ويمكنك الاطّلاع على مزيد من التفاصيل حول هذا النمط في مقالات المدوّنة التالية:

• التقاط بيانات التغيير (CDC) باستخدام PostgreSQL وClickHouse - الجزء 1
• التقاط بيانات التغيير (CDC) باستخدام PostgreSQL وClickHouse - الجزء 2

​

1. إعداد دفق Kinesis

أولاً، ستحتاج إلى تمكين دفق Kinesis على جدول DynamoDB لالتقاط التغييرات في الوقت الفعلي. نريد تنفيذ ذلك قبل إنشاء اللقطة لتجنّب فقدان أي بيانات. يمكنك العثور على دليل AWS هنا.

​

2. إنشاء اللقطة

بعد ذلك، سننشئ لقطة من جدول DynamoDB. ويمكن تحقيق ذلك عبر تصدير من AWS إلى S3. يمكنك الاطلاع على دليل AWS هنا. ستحتاج إلى إجراء “Full export” بتنسيق DynamoDB JSON.

​

3. تحميل اللقطة إلى ClickHouse

​

أنشئ الجداول اللازمة

ستبدو بيانات اللقطة من DynamoDB على النحو التالي:

{
  "age": {
    "N": "26"
  },
  "first_name": {
    "S": "sally"
  },
  "id": {
    "S": "0A556908-F72B-4BE6-9048-9E60715358D4"
  }
}

لاحظ أن البيانات بتنسيق متداخل. سنحتاج إلى تسطيح هذه البيانات قبل تحميلها إلى ClickHouse. ويمكن إجراء ذلك باستخدام الدالة JSONExtract في ClickHouse داخل عرض مادي.

سنحتاج إلى إنشاء ثلاثة جداول:

1. جدول لتخزين البيانات الخام من DynamoDB
2. جدول لتخزين البيانات النهائية بعد تسطيحها (جدول الوجهة)
3. عرض مادي لتسطيح البيانات

بالنسبة إلى بيانات DynamoDB الواردة في المثال أعلاه، ستبدو جداول ClickHouse على النحو التالي:

/* Snapshot table */
CREATE TABLE IF NOT EXISTS "default"."snapshot"
(
    `item` String
)
ORDER BY tuple();

/* Table for final flattened data */
CREATE MATERIALIZED VIEW IF NOT EXISTS "default"."snapshot_mv" TO "default"."destination" AS
SELECT
    JSONExtractString(item, 'id', 'S') AS id,
    JSONExtractInt(item, 'age', 'N') AS age,
    JSONExtractString(item, 'first_name', 'S') AS first_name
FROM "default"."snapshot";

/* Table for final flattened data */
CREATE TABLE IF NOT EXISTS "default"."destination" (
    "id" String,
    "first_name" String,
    "age" Int8,
    "version" Int64
)
ENGINE ReplacingMergeTree("version")
ORDER BY id;

توجد بعض المتطلبات لجدول الوجهة:

• يجب أن يكون هذا الجدول من نوع ReplacingMergeTree
• يجب أن يحتوي الجدول على عمود version
  • في الخطوات اللاحقة، سنربط الحقل ApproximateCreationDateTime من Kinesis Stream بالعمود version.
• ينبغي أن يستخدم الجدول مفتاح التقسيم بوصفه مفتاح الترتيب (المحدَّد بواسطة ORDER BY)
  • ستُزال تكرارات الصفوف ذات مفتاح الترتيب نفسه استنادًا إلى العمود version.

​

أنشئ ClickPipe للّقطة

يمكنك الآن إنشاء ClickPipe لتحميل بيانات اللقطة من S3 إلى ClickHouse. اتبع دليل S3 ClickPipe هنا، ولكن استخدم الإعدادات التالية:

• مسار الاستيعاب: ستحتاج إلى تحديد مسار ملفات JSON المُصدَّرة في S3. سيبدو المسار على النحو التالي تقريبًا:

https://{bucket}.s3.amazonaws.com/{prefix}/AWSDynamoDB/{export-id}/data/*

• التنسيق: JSONEachRow
• الجدول: جدول اللقطة الخاص بك (مثل default.snapshot في المثال أعلاه)

بمجرد إنشائه، ستبدأ البيانات بالتدفق إلى جدول اللقطة وجدول الوجهة. لا تحتاج إلى انتظار اكتمال تحميل اللقطة قبل الانتقال إلى الخطوة التالية.

​

4. إنشاء Kinesis ClickPipe

يمكننا الآن إعداد Kinesis ClickPipe لالتقاط التغييرات الآنية من دفق Kinesis. اتبع دليل Kinesis ClickPipe هنا، ولكن استخدم الإعدادات التالية:

• Stream: دفق Kinesis المستخدم في الخطوة 1
• Table: جدول الوجهة (مثل default.destination في المثال أعلاه)
• Flatten object: true
• Column mappings:
  • ApproximateCreationDateTime: version
  • عيّن الحقول الأخرى إلى أعمدة الوجهة المناسبة كما هو موضح أدناه

​

5. التنظيف (اختياري)

بعد انتهاء ClickPipe الخاصة باللقطة، يمكنك حذف جدول اللقطة والعرض المادي.

DROP TABLE IF EXISTS "default"."snapshot";
DROP TABLE IF EXISTS "default"."snapshot_clickpipes_error";
DROP VIEW IF EXISTS "default"."snapshot_mv";