كيفية تنفيذ تجميعات Bucket في ElasticSearch على حقول قد تكون بياناتها مفقودة

Elasticsearch هو محرك بحث وتحليلات معروف بسرعة استجابته الكبيرة وطريقته السهلة لتنفيذ عمليات CRUD على البيانات، وهو ما تتيحه واجهات REST البرمجية. Elasticsearch جزء من Elastic Stack الذي يضم أداتين إضافيتين، Logstash وKibana.

يُستخدَم ElasticSearch عادةً في نمط CQRS (فصل مسؤولية الأوامر والاستعلامات) حيث نفصل عمليات القراءة والتحديث. يُستخدَم هذا النهج لزيادة الأداء وقابلية التوسع والأمان.

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

كانت المهمة تفعيل تصفية المستخدمين بناءً على الخبرة التي يمتلكها المستخدم في مسيرته المهنية. لفهم ذلك بشكل أفضل، نحتاج لإلقاء نظرة على بنية خاصية experience الخاصة بالمستخدم داخل فهرسنا الرئيسي.

{
  "_index": "...",
  "_type": "_doc",
  "_id": "...",
  "_score": 1.0,
  "_source": {
    ...
    "experiences": [
      {
        "title": "Software Engineer",
        "company": "Valens Dev",
        "startDate": "2022-07-12T18:20:17.439Z",
        "description": "string",
        "isEmployed": true,
        "location": {
          "name": "Sarajevo"
        },
        "contractType": {
          "name": "...",
          "externalId": null
        },
        "endDate": null,
        "externalId": null,
      }
    ],
  ...
  }
}

0.1 - مثال على مستند فهرس أساسي (خاصية experience داخل الفهرس الأساسي)

في المستند أعلاه يمكننا رؤية خاصية experience التي تعطينا فكرة عن كيفية معالجة المشكلة. الآن بعد أن حددنا المشكلة ورأينا بنية المستند، سنحدد الحل:

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

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

بهذه الطريقة تتجنب مشاكل كيفية الوصول إلى البيانات في الخصائص المتداخلة.

هذا مستند من فهرس experience الذي أنشأته إلى جانب الفهرس الرئيسي، هنا أحفظ فقط البيانات الضرورية اللازمة لتنفيذ التجميعات:

{
  {
    "id": 14214,
    "userId": 135151,
    "startDate": "2022-07-12T18:20:17.439Z",
    "endDate": null,
    "isEmployed": true
  }
}

0.2 - مستند داخل فهرس Experience

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

{
  "aggregations": {
    "group_by": {
      "terms": {
        "field": "userId"
      }
    }
  }
}

0.3 - تجميع مستندات experience حسب userId الخاص بها

كما ترى في مثال الكود أعلاه، نُعلن عن خاصية aggregation ثم نحدد terms التي نريد group_by حسبها، في حالتنا نجمّع مستنداتنا حسب الحقل userId المخزَّن في فهرسنا.

{
  "aggregations": {
    "group_by": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": 101,
          "doc_count": 1
        }
      ]
    }
  }
}

0.4 - مستندات مُجمَّعة حسب معرّفاتها ومقسَّمة إلى buckets

لدينا الآن نتيجة تجميعنا الأول، الجزء الأهم هنا هو رؤية خاصية buckets، وهي مصفوفة من كائنات bucket تحتوي على key وهو userId لدينا، وخاصية doc_count وهي عدد المستندات التي تحمل نفس userId.

سيتألف هذا القسم من جزأين، سيصف الجزء الأول تجميع sum لحقلي تاريخ البداية والنهاية، والجزء الثاني سيأخذ هذين الحقلين ويطرحهما ويحصل على عدد الأشهر لمستخدم معين.

{
  "aggregations": {
    "start": {
      "sum": {
        "field": "startDate"
      }
    },
    "end": {
      "sum": {
        "field": "endDate"
      }
    }
  }
}

0.5 - جمع كل تواريخ startDate وendDate

كما رأينا في المثال السابق، نبدأ التجميع بالإعلان عن خاصية aggregation والآن بما أننا نجري تجميع sum نريد الإعلان عن خاصية ستُعرَض في bucket لدينا عند اكتمال التجميع.

سيحمل مجموع حقول startDate لدينا اسم الخاصية start، ومجموع حقول endDate لدينا سيحمل اسم الخاصية end.

ملاحظة أن هذا التجميع سيكون متداخلًا تحت خاصية group_by الخاصة بالتجميع الأول الذي أجريناه في المثال أعلاه، يمكنك أيضًا متابعة الاستعلام الكامل الموجود في نهاية هذا المقال.

{
  "buckets": [
    {
      "key": 101,
      "doc_count": 1,
      "start": {
        "value": 1.6451424e12,
        "value_as_string": "2022-02-18T00:00:00.000Z"
      },
      "end": {
        "value": 1.6660512e12,
        "value_as_string": "2022-10-18T00:00:00.000Z"
      }
    }
  ]
}

0.6 - نتيجة تجميع Sum

هذه نتيجة تجميعنا، والآن يحتوي bucket لدينا على خاصيتين جديدتين هما start وend اللتين حددناهما في مثال الكود 0.5 وتحتويان أيضًا على خصائص لتواريخنا. ستكون خاصية value مفيدة بشكل خاص في الجزء التالي...

{
  "duration": {
    "bucket_script": {
      "buckets_path": {
        "start": "start.value",
        "end": "end.value"
      },
      "script": {
        "params": {
          "month_in_milliseconds": 2628000000
        },
        "source": "Math.round((params.end - params.start) / params.month_in_milliseconds)"
      }
    }
  }
}

0.7 - تجميع Bucket

الآن بعد أن جمعنا تواريخ البداية والنهاية، نحتاج لطرح تاريخ النهاية من تاريخ البداية للحصول على عدد الأشهر التي كان فيها المستخدم موظَّفًا. ستكون نتيجة تلك العملية بالميلي ثانية، ولتحويل الميلي ثانية إلى أعداد صحيحة فعلية، سنطبّق دالة Math.round() في Java.

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

• الخطوة الأولى هي إنشاء سكريبت bucket وتسميته بشكل مناسب، في حالتنا نحاول إيجاد إجمالي مدة التوظيف لمستخدم معين، لذا سنسمي bucket لدينا duration.

• ثم في الخطوة الثانية سنحدد الحقل duration كـ bucket_script الذي سيخبر Elastic بأن يعتبره تجميع bucket.

• داخل سكريبت الـ bucket لدينا خاصية bucketpath التي تتيح لنا تحديد الخصائص التي نريد استخدامها في تجميع الـ bucket لدينا، لكن يجب أن تكون تلك الخصائص موجودة في الـ bucket. للتأكد من أننا نستخدم الخصائص الموجودة في bucket لدينا، سنلقي نظرة على مقتطف الكود 0.6 وهناك نرى خاصيتي start وend، واللتين تحملان كلتاهما خاصية value التي سنستخدمها في الجزء التالي من هذا التجميع. داخل خاصية bucketspath، نحدد خاصيتي start وend اللتين ستشيران إلى خصائص bucket تجميع sum وهما start.value وend.value.

• ثم نحدد خاصية script، وهذا هو الجزء الرئيسي لأن كل السحر يحدث هنا، هنا نحدد params التي يمكن أن تحتوي على أي شيء نريده وهي الطريقة المفضَّلة لتحديد متغيرات ثابتة ستُستخدَم في خاصية source لتنفيذ العمليات.

• وفي النهاية خاصية source. سيحتوي حقل duration في bucket على القيمة المحسوبة في هذه الخاصية، وفي حالتنا كما وُصف سابقًا، نطرح خاصية end من خاصية start ثم نقسّمها على المتغيّر monthinmilliseconds وبعدها ستُقرَّب النتيجة باستخدام دالة Math.round() في Java.

ملاحظة سيكون هذا التجميع أيضًا متداخلًا داخل التجميع الثاني حيث جمعنا حقلي تاريخ البداية والنهاية (مقتطف الكود 0.5)

{
  "buckets": [
    {
      "key": 101,
      "doc_count": 1,
      "start": {
        "value": 1.6451424e12,
        "value_as_string": "2022-02-18T00:00:00.000Z"
      },
      "end": {
        "value": 1.6660512e12,
        "value_as_string": "2022-10-18T00:00:00.000Z"
      },
      "duration": {
        "value": 8.0
      }
    }
  ]
}

0.8 - نتيجة تجميع bucket الموصوف في مقتطف الكود 0.7

وكما نرى، تحتوي خاصية duration على الرقم 8 وهو الفرق بين قيمتي end وstart، إذن يمتلك مستخدمنا 8 أشهر من الخبرة وهذا صحيح.

للتأكد من أن نتيجة التجميع صحيحة، يمكنك النظر إلى التمثيل النصي في خاصية valueasstring لكل من start وend وحساب الفرق بينهما يدويًا.

في نهاية هذه العملية نريد تصفية bucket الخاص بـ duration لإعادة فقط المستخدمين الذين يطابقون الشرط المُمرَّر. تشبه هذه العملية العملية أعلاه لكننا هنا فقط نصفّي الـ bucket الذي تم تجميعه وسنفعل ذلك بالصياغة التالية التي سأشرحها خطوة بخطوة.

{
  "duration_bucket_filter": {
    "bucket_selector": {
      "buckets_path": {
        "durationBucket": "duration"
      },
      "script": {
        "params": {
          "min_number_of_months": 8,
          "max_number_of_months": 8
        },
        "source": "params.durationBucket >= params.min_number_of_months && params.durationBucket <= params.max_number_of_months"
      }
    }
  }
}

0.9 - تصفية bucket الخاص بـ duration باستخدام مرشّح bucket الخاص بالمدة

بالنظر إلى مثال الكود أعلاه في مقتطف الكود 0.9 يمكنك رؤية نمط يظهر عندما يتعلق الأمر بتجميعات bucket. نحدد أولًا الاسم، في هذه الحالة durationbucketfilter، ثم نحدد نوع تجميع الـ bucket، في المثال 0.7 كان لدينا bucketscript المُستخدَم لتجميع البيانات، والآن بدلًا من سكريبت bucket نحدد bucketselector الذي يخبر Elastic بأننا نريد اختيار bucket.

لاختيار خاصية داخل bucket كما في مقتطف الكود 0.7 نضيف خاصية buckets_path حيث نحدد durationBucket كاسم مستعار لخاصية duration التي جُمِّعت في الخطوة السابقة (مقتطف الكود 0.8).

الآن بعد تحديد الخاصية، نريد تصفية buckets لدينا. بإضافة خاصية scripts نحدد params التي نريد استخدامها داخل source لدينا. هنا حددنا الحد الأدنى والأقصى لعدد الأشهر التي يحتاجها المستخدم حتى تُعاد bucket الخاصة به بعد التصفية.

وفي نهاية مقتطف الكود هذا خاصية source التي تحتوي على منطق تصفية الـ buckets:

• params.durationBucket >= params.minnumberof_months - يجب أن تكون مدة خبرة معينة أكبر من الحد الأدنى لعدد الأشهر
• params.durationBucket <= params.maxnumberof_months - يجب أن تكون تلك الخبرة أقل من أو تساوي الحد الأقصى لعدد الأشهر

بناءً على هذه الشروط تُصفَّى الـ buckets وتُعاد:

{
  "buckets": [
    {
      "key": 101,
      "doc_count": 1,
      "start": {
        "value": 1.6451424e12,
        "value_as_string": "2022-02-18T00:00:00.000Z"
      },
      "end": {
        "value": 1.6660512e12,
        "value_as_string": "2022-10-18T00:00:00.000Z"
      },
      "duration": {
        "value": 8.0
      }
    }
  ]
}

1.0 - نتيجة تصفية bucket

كما ترى، ضبطنا القيمتين الدنيا والقصوى لمطابقة المستخدم بالمفتاح 101 للحصول فقط على ذلك المستخدم في نتيجة تصفية bucket لدينا.

"runtime_mappings": {
  "updatedEndDate": {
    "type": "date",
    "script": {
      "source": "if (doc['isEmployed'].value.equals(true)) { emit(new Date().getTime()) } else { emit(doc['endDate'].value.millis) }"
    }
  }
}

1.1 - تعبئة حقل updatedEndDate

الآن أوشكنا على الانتهاء لكن هناك أمر إضافي واحد، ماذا لو كان المستخدم لا يزال موظَّفًا؟ كيف نحسب الفرق بين تاريخ البداية والنهاية إذا كان تاريخ النهاية null؟ الإجابة بسيطة جدًا: حقول runtime المدعومة ابتداءً من الإصدار 7.11 لذا تأكد من امتلاك إصدار Elastic المناسب.

تتيح لنا حقول runtime تعبئة الحقول أثناء التنفيذ (runtime). هذا يعني أن القيمة تُولَّد عند الحاجة إليها، في حالتنا عندما يقوم شخص ما بتصفية المستخدمين الذين لديهم بعض سنوات الخبرة ولا يزالون موظَّفين، سيُملأ حقل endDate بدلًا من null بالتاريخ عند لحظة الاستعلام عن البيانات.

لتنفيذ حقول runtime في استعلامنا، سنضيف خاصية runtimemappings في أعلى الاستعلام. ثم سنضيف لها نوعًا (type) وهو date في حالتنا، ثم نحدد خاصية script يليها المصدر حيث تؤدي runtimemappings سحرها، سنستعرض خاصية source الآن حتى تفهمها وتكيّفها حسب احتياجاتك:

• أولًا نتحقق مما إذا كان المستخدم موظَّفًا if (doc['isEmployed'].value.equals(true)) عبر الحقل isEmployed الموجود في كل مستند من فهرسنا
• إذا كان موظَّفًا فهذا يعني أن حقل endDate هو null وأننا نحتاج لتعبئته بالتاريخ الحالي. لفعل ذلك نُصدر (emit) قيمة التاريخ الحالي إلى الحقل الجديد المُسمى updatedEndDate بجزء الكود المعروض هنا { emit(new Date().getTime()) }

• إذا لم يكن موظَّفًا، فهذا يعني أن حقل endDate مُعبَّأ فعليًا ونحتاج لإصداره في الحقل الجديد updatedEndDate
• ستتم هذه العملية بالكود التالي { emit(doc['endDate'].value.millis) }، نُصدر التاريخ بالميلي ثانية.

الآن جمّعنا استعلامنا وقدّمنا حلًا للمشكلة الموصوفة في بداية المقال. يمكنك الاطلاع على الاستعلام الكامل أدناه في مقتطف الكود 1.2

"runtime_mappings": {
  "updatedEndDate": {
    "type": "date",
    "script": {
      "source": "if (doc['isEmployed'].value.equals(true)) { emit(new Date().getTime()) } else { emit(doc['endDate'].value.millis) }"
    }
  }
},
"aggregations": {
  "group_by": {
    "terms": {
      "field": "userId"
    },
    "aggregations": {
      "start": {
        "sum": {
          "field": "startDate"
        }
      },
      "end": {
        "sum": {
          "field": "updatedEndDate"
        }
      },
      "duration": {
        "bucket_script": {
          "buckets_path": {
            "start": "start.value",
            "end": "end.value"
          },
          "script": {
            "params": {
              "month_in_milliseconds": 2628000000
            }
            "source": "Math.round((params.end - params.start) / params.month_in_milliseconds)"
          }
        }
      },
      "duration_bucket_filter": {
        "bucket_selector": {
          "buckets_path": {
            "durationBucket": "duration"
          },
          "script": {
            "params": {
              "min_number_of_months": 8,
              "max_number_of_months": 8
            },
            "source": "params.durationBucket >= params.min_number_of_months && params.durationBucket <= params.max_number_of_months"
          }
        }
      }
    }
  }
}

1.2 - مقتطف الكود الكامل

• Elasticsearch
• CQRS
• تجميع Terms
• المدى بين تاريخين
• Elasticsearch Transforms: حساب إجمالي المدة من تحديثات حالة متعددة
• ElasticSearch - الفرق بين حقلي تاريخ
• تحويل التاريخ إلى ميلي ثانية في elasticsearch
• التحقق من القيمة الافتراضية في elastic
• Runtime mappings
• Elasticsearch – تجميع Bucket Selector