محرك الجدول Distributed

الجداول التي تستخدم المحرك Distributed لا تخزّن أي بيانات خاصة بها، لكنها تتيح معالجة الاستعلامات الموزعة عبر عدة خوادم. تُنفَّذ القراءة بالتوازي تلقائيًا. أثناء القراءة، تُستخدم فهارس الجداول على الخوادم البعيدة إن وُجدت.

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]])
[SETTINGS name=value, ...]

عندما يشير جدول Distributed إلى جدول على الخادم الحالي، يمكنك اعتماد مخطط هذا الجدول:

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] AS [db2.]name2 ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]]) [SETTINGS name=value, ...]

يُعد Remote وRemoteSecure محركَي جداول دائمة، ويقبلان تعبيرات العناوين نفسها وبيانات الاعتماد نفسها التي تقبلها دوال الجداول remote وremoteSecure:

CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
    name1 [type1],
    name2 [type2],
    ...
) ENGINE = Remote(addresses_expr, [db, table, [user [, password], sharding_key]])
[SETTINGS name = value, ...]

يقبل RemoteSecure الوسيطات نفسها ويتصل عبر اتصال آمن (ويُستخدم منفذ TCP الآمن افتراضيًا). وتُفسَّر الوسيطات بالطريقة نفسها تمامًا كما في دوال الجدول remote وremoteSecure؛ راجع الوصف الخاص بهما لمعرفة التواقيع المدعومة. ويمكن حذف بنية الجدول، وفي هذه الحالة يُستنتج من الجدول البعيد.

تُحدَّد إعدادات وحدة التخزين المُنشأة، مثل skip_unavailable_shards، بعد تعريف المحرك، على سبيل المثال ENGINE = Remote('127.0.0.1', system, one) SETTINGS skip_unavailable_shards = 1. لاحظ أن دالتي الجدول remote وremoteSecure تقبلان بدلًا من ذلك عبارة SETTINGS ضمن وسيطاتهما، remote('127.0.0.1', system.one, SETTINGS skip_unavailable_shards = 1)، إذ لا يوجد موضع آخر لوضعها في دالة الجدول؛ أما المحركات فلا تقبل هذا الشكل.

على سبيل المثال:

CREATE TABLE remote_one ENGINE = Remote('127.0.0.1', system, one);
SELECT * FROM remote_one;

هذا هو المقابل الدائم لـ CREATE TABLE ... AS remote(...). وكما هو الحال مع دوال الجداول ‏remote، فإن هذه المحركات عملية، لكنها لا تتيح لك إعداد الشاردات والنسخ المتماثلة بصورة تعريفية كما يفعل Distributed فوق عنقود مُهيّأ، لذا إذا كانت لديك مجموعة دائمة من الخوادم تُستخدَم كثيرًا، فمن الأفضل تعريف عنقود واستخدام المحرك Distributed.

يمكن أن يكون الهدف أيضًا دوال الجداول، على سبيل المثال Remote('127.0.0.1', numbers(10)) أو Remote('127.0.0.1', merge(db, '^table_')). يكون هذا الجدول للقراءة فقط: إذ لا يوجد جدول بعيد يمكن الإدراج فيه، لذلك يُرفَض INSERT مع الاستثناء NOT_IMPLEMENTED. وينطبق هذا القيد الخاص بالقراءة فقط أيضًا على دوال الجداول ‏remote وremoteSecure: إذ إن كلًا من SELECT وINSERT مدعومان عند استخدام هدف db/table عادي، لكن الهدف من نوع دوال الجداول ‏(remote('127.0.0.1', numbers(10))) يكون للقراءة فقط للسبب نفسه.

انظر أيضًا

• الإعداد distributed_foreground_insert
• MergeTree للاطلاع على الأمثلة

مثال

CREATE TABLE hits_all AS hits
ENGINE = Distributed(logs, default, hits[, sharding_key[, policy_name]])
SETTINGS
    fsync_after_insert=0,
    fsync_directories=0;

ستُقرأ البيانات من جميع الخوادم في عنقود logs، من جدول default.hits الموجود على كل خادم في العنقود. ولا يقتصر الأمر على قراءة البيانات فحسب، بل تُعالَج أيضًا جزئيًا على الخوادم البعيدة (بالقدر الممكن). على سبيل المثال، في استعلام يتضمن GROUP BY، ستُجمَّع البيانات على الخوادم البعيدة، وستُرسَل الحالات الوسيطة للدوال التجميعية إلى الخادم الطالب. ثم تُجمَّع البيانات مرةً أخرى.

بدلًا من اسم قاعدة البيانات، يمكنك استخدام تعبير ثابت يُرجع سلسلة نصية. على سبيل المثال: currentDatabase().

تُضبط العناقيد في ملف إعدادات الخادم:

<remote_servers>
    <logs>
        <!-- Inter-server per-cluster secret for Distributed queries
             default: no secret (no authentication will be performed)

             If set, then Distributed queries will be validated on shards, so at least:
             - such cluster should exist on the shard,
             - such cluster should have the same secret.

             And also (and which is more important), the initial_user will
             be used as current user for the query.
        -->
        <!-- <secret></secret> -->

        <!-- Optional. Whether distributed DDL queries (ON CLUSTER clause) are allowed for this cluster. Default: true (allowed). -->
        <!-- <allow_distributed_ddl_queries>true</allow_distributed_ddl_queries> -->

        <shard>
            <!-- Optional. Shard weight when writing data. Default: 1. -->
            <weight>1</weight>
            <!-- Optional. The shard name.  Must be non-empty and unique among shards in the cluster. If not specified, will be empty. -->
            <name>shard_01</name>
            <!-- Optional. Whether to write data to just one of the replicas. Default: false (write data to all replicas). -->
            <internal_replication>false</internal_replication>
            <replica>
                <!-- Optional. Priority of the replica for load balancing (see also load_balancing setting). Default: 1 (less value has more priority). -->
                <priority>1</priority>
                <host>example01-01-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-01-2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <weight>2</weight>
            <name>shard_02</name>
            <internal_replication>false</internal_replication>
            <replica>
                <host>example01-02-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-02-2</host>
                <secure>1</secure>
                <port>9440</port>
            </replica>
        </shard>
    </logs>
</remote_servers>

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

يجب ألا تحتوي أسماء العناقيد على نقاط.

تُحدَّد المعلمات host وport، واختياريًا user وpassword وsecure وcompression وbind_host لكل خادم:

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

يمكنك تحديد شارد واحد فقط (وفي هذه الحالة ينبغي أن تُسمّى معالجة الاستعلام remote بدلًا من distributed)، أو أي عدد من الشوارد. وفي كل شارد، يمكنك تحديد نسخة متماثلة واحدة أو أي عدد من النسخ المتماثلة. ويمكنك تحديد عدد مختلف من النسخ المتماثلة لكل شارد.

يمكنك تحديد أي عدد تريده من العناقيد في التهيئة.

لعرض عناقيدك، استخدم الجدول system.clusters.

يتيح المحرك Distributed العمل مع العنقود كما لو كان خادمًا محليًا. ومع ذلك، لا يمكن تحديد تهيئة العنقود ديناميكيًا، بل يجب ضبطها في ملف إعدادات الخادم. وعادةً ما تكون لدى جميع الخوادم في العنقود تهيئة العنقود نفسها (مع أن هذا ليس مطلوبًا). وتُحدَّث العناقيد من ملف الإعدادات أثناء التشغيل، من دون إعادة تشغيل الخادم.

إذا كنت بحاجة إلى إرسال استعلام إلى مجموعة غير معروفة من الشوارد والنسخ المتماثلة في كل مرة، فلست بحاجة إلى إنشاء جدول Distributed — استخدم بدلًا من ذلك دالة الجدول remote. راجع قسم دوال الجداول.

هناك طريقتان لكتابة البيانات إلى cluster:

أولًا، يمكنك تحديد الخوادم التي ستُكتب إليها البيانات المطلوبة، ثم تنفيذ الكتابة مباشرةً على كل shard. وبعبارة أخرى، نفّذ عبارات INSERT مباشرةً على الجداول البعيدة في cluster التي يشير إليها جدول Distributed. هذا هو الحل الأكثر مرونة، إذ يمكنك استخدام أي مخطط sharding، حتى لو لم يكن بسيطًا بسبب متطلبات مجال العمل. وهو أيضًا الحل الأمثل، لأن البيانات يمكن كتابتها إلى shards مختلفة بشكل مستقل تمامًا.

ثانيًا، يمكنك تنفيذ عبارات INSERT على جدول Distributed. في هذه الحالة، سيتولى الجدول بنفسه توزيع البيانات المُدرجة على الخوادم. ولكي تتمكن من الكتابة إلى جدول Distributed، يجب أن يكون المعامل sharding_key مُعدًّا (إلا إذا كان هناك shard واحد فقط).

يمكن تعريف <weight> لكل shard في config file. وتكون القيمة الافتراضية للوزن هي 1. تُوزَّع البيانات على shards بكميات تتناسب مع وزن كل shard. وتُجمع أوزان جميع shards، ثم يُقسم وزن كل shard على المجموع الكلي لتحديد نسبة كل shard. على سبيل المثال، إذا كان هناك shardان وكان وزن الأول 1 ووزن الثاني 2، فسيُرسل ثلث الصفوف المُدرجة (1 / 3) إلى الأول، ويُرسل الثلثان (2 / 3) إلى الثاني.

يمكن تعريف المعامل internal_replication لكل shard في config file. إذا ضُبط هذا المعامل على true، فستختار عملية الكتابة أول replica سليمة وتكتب البيانات إليها. استخدم هذا إذا كانت الجداول الأساسية التي يستند إليها جدول Distributed هي replicated tables (مثل أي من table engines من النوع Replicated*MergeTree). ستتلقى إحدى replicas الخاصة بالجدول عملية الكتابة، ثم ستُنسخ البيانات تلقائيًا إلى replicas الأخرى.

إذا ضُبط internal_replication على false (وهو الإعداد الافتراضي)، فستُكتب البيانات إلى جميع replicas. في هذه الحالة، يتولى جدول Distributed نفسه replication البيانات. وهذا أسوأ من استخدام replicated tables لأن اتساق replicas لا يُتحقق منه، ومع مرور الوقت ستحتوي على بيانات مختلفة قليلًا.

لاختيار shard الذي يُرسل إليه صف من البيانات، يُحلَّل تعبير sharding، ثم يُؤخذ remainder ناتج قسمته على إجمالي وزن shards. ويُرسل الصف إلى shard الذي يقابل نصف المجال من البواقي من prev_weights إلى prev_weights + weight، حيث إن prev_weights هو إجمالي وزن shards ذات الأرقام الأصغر، وweight هو وزن هذا shard. على سبيل المثال، إذا كان هناك shardان، وكان وزن الأول 9 ووزن الثاني 10، فسيُرسل الصف إلى shard الأول إذا كان الباقي ضمن المجال [0, 9)، وإلى الثاني إذا كان الباقي ضمن المجال [9, 19).

يمكن أن يكون تعبير sharding أي expression مكوَّن من ثوابت وcolumns الجدول ويُرجع عددًا صحيحًا. على سبيل المثال، يمكنك استخدام التعبير rand() للتوزيع العشوائي للبيانات، أو UserID للتوزيع بحسب باقي قسمة معرّف المستخدم (وعندئذٍ ستوجد بيانات المستخدم الواحد على shard واحد، مما يسهّل تنفيذ IN وJOIN بحسب المستخدمين). وإذا لم يكن أحد columns موزعًا بالتساوي بما يكفي، يمكنك تغليفه بـ hash function مثل intHash64(UserID).

إن الاعتماد على remainder القسمة وحده حل محدود لـ sharding، وليس مناسبًا دائمًا. فهو يعمل مع أحجام البيانات المتوسطة والكبيرة (عشرات الخوادم)، لكنه لا يناسب أحجام البيانات الضخمة جدًا (مئات الخوادم أو أكثر). وفي الحالة الأخيرة، استخدم مخطط sharding الذي يتطلبه مجال العمل بدلًا من استخدام entries في جداول Distributed.

ينبغي أن تهتم بمخطط sharding في الحالات التالية:

• تُستخدم queries تتطلب join البيانات (IN أو JOIN) باستخدام مفتاح محدد. إذا كانت البيانات مُجزأة وفق هذا المفتاح، فيمكنك استخدام IN أو JOIN المحليين بدلًا من GLOBAL IN أو GLOBAL JOIN، وهذا أكثر كفاءة بكثير.
• يُستخدم عدد كبير من الخوادم (مئات أو أكثر) مع عدد كبير من queries الصغيرة، مثل queries الخاصة ببيانات عملاء منفردين (مثل مواقع الويب أو المعلنين أو الشركاء). ولكي لا تؤثر queries الصغيرة في cluster بأكمله، فمن المنطقي وضع بيانات العميل الواحد على shard واحد. وبدلًا من ذلك، يمكنك إعداد sharding ثنائي المستوى: تقسيم cluster بأكمله إلى "طبقات"، بحيث قد تتكون الطبقة من عدة shards. توضع بيانات العميل الواحد على طبقة واحدة، لكن يمكن إضافة shards إلى الطبقة عند الحاجة، وتُوزَّع البيانات عشوائيًا داخلها. تُنشأ جداول Distributed لكل طبقة، ويُنشأ جدول distributed مشترك واحد للاستعلامات العامة.

تُكتَب البيانات في الخلفية. عند إدراجها في الجدول، لا تُكتَب كتلة البيانات إلا إلى نظام الملفات المحلي. وتُرسَل البيانات إلى الخوادم البعيدة في الخلفية في أقرب وقت ممكن. ويُتحكَّم في دورية إرسال البيانات بواسطة الإعدادَين distributed_background_insert_sleep_time_ms وdistributed_background_insert_max_sleep_time_ms. يرسل محرك Distributed كل ملف يحتوي على بيانات مُدرَجة بشكل منفصل، ولكن يمكنك تفعيل الإرسال الدفعي للملفات باستخدام الإعداد distributed_background_insert_batch. يُحسِّن هذا الإعداد أداء العنقود من خلال الاستفادة على نحو أفضل من موارد الخادم المحلي والشبكة. ينبغي التحقق من إرسال البيانات بنجاح عبر فحص قائمة الملفات (البيانات التي تنتظر الإرسال) في دليل الجدول: /var/lib/clickhouse/data/database/table/. ويمكن ضبط عدد الخيوط التي تنفّذ المهام في الخلفية بواسطة الإعداد background_distributed_schedule_pool_size.

إذا تعطل الخادم أو تعرّض لإعادة تشغيل غير سليمة (على سبيل المثال، بسبب عطل في العتاد) بعد تنفيذ INSERT إلى جدول Distributed، فقد تُفقَد البيانات المُدرَجة. وإذا اكتُشف جزء بيانات تالف في دليل الجدول، فسيُنقَل إلى الدليل الفرعي broken ولن يُستخدَم بعد ذلك.

عند الاستعلام عن جدول Distributed، تُرسَل استعلامات SELECT إلى جميع الشوارد وتعمل بغضّ النظر عن كيفية توزيع البيانات بينها (إذ يمكن أن تكون موزعة عشوائيًا بالكامل). وعند إضافة شارد جديد، لا تحتاج إلى نقل البيانات القديمة إليه. بدلًا من ذلك، يمكنك كتابة البيانات الجديدة إليه باستخدام وزن أعلى — ستتوزع البيانات بشكل غير متساوٍ قليلًا، لكن الاستعلامات ستعمل بكفاءة وبصورة صحيحة.

عند تمكين الخيار max_parallel_replicas، تتم موازاة معالجة الاستعلام عبر جميع النسخ المتماثلة داخل شارد واحد. لمزيد من المعلومات، راجع قسم max_parallel_replicas.

لمعرفة المزيد حول كيفية معالجة الاستعلامات الموزعة in وglobal in، راجع هذه الوثائق.

_shard_num — يحتوي على قيمة shard_num من الجدول system.clusters. النوع: UInt32.

انظر أيضًا

• وصف الأعمدة الافتراضية
• الإعداد background_distributed_schedule_pool_size
• الدالتان shardNum() وshardCount()