مراقبة Apache Spark: كيفية استخدام واجهة برمجة تطبيقات Spark والمكتبات مفتوحة المصدر لتحسين قابلية ملاحظة البيانات في تطبيقاتك

يُعد Spark مكوّناً محورياً في مجموعة البيانات الحديثة. على هذا النحو، من المهم للغاية توفير مستوى مناسب من قابلية الملاحظة لبيئات Spark لديك. توجد العديد من خيارات مراقبة Spark، بما في ذلك برامج SaaS التي توفّر لوحات معلومات جاهزة لمقاييس Spark وSpark SQL. لكن ماذا لو لم يكن ذلك كافياً؟

يشتمل إعداد تطبيق Spark النموذجي، سواء كان مُستضافاً ذاتياً أو مُداراً، على عدد من لوحات المعلومات التشغيلية لمراقبة سلامة المجموعات. ورغم فائدة هذه اللوحات، فإنها لا تقدّم سوى عرض عام للبنية التحتية، ولا تُظهر المقاييس الفعلية المرتبطة بالبيانات نفسها. قد يشير ارتفاع استهلاك وحدة المعالجة المركزية أو اقتراب مجموعة Spark من استنفاد الذاكرة إلى وجود مشكلة في التطبيق، لكن هذا النوع من المؤشرات لا يفيد عندما يكون مصدر البيانات قد غيّر مخطط البيانات (schema) أو كانت البيانات الواردة من قسم آخر تالفة. معظم المشكلات التي يواجهها المهندسون تعود إلى البيانات لا إلى البنية التحتية؛ لذا يضطرّون إلى قضاء وقت طويل في إعادة إنتاج المشكلة أو التنقّل بين الملفات وحاويات التخزين (buckets) وكأنهم يحققون في قضية. وهنا تبرز أهمية مراقبة التطبيق نفسها.

كل حالة تتطلّب مستوى مختلفاً من الرؤية، ويحتاج مهندسو البيانات إلى القدرة على التعمّق أبعد من مجرد مقاييس التنفيذ. وإلا فقد تقضي وقتاً طويلاً في تتبّع مشكلات جودة البيانات وتصحيحها في Spark.

في هذا الدليل، ستتعرّف على كيفية تحقيق مستويات مختلفة من إمكانية ملاحظة البيانات في Spark، سواء على مستوى الصورة العامة أو على مستوى التفاصيل الدقيقة. على مستوى الصورة العالية، ستعتمد على الأنظمة الداخلية في Spark، مثل واجهات Listener APIs وQuery Execution Listeners. أما على المستوى التفصيلي، فستتعلّم كيف تستخدم المكتبات لتتبّع مقاييس جودة البيانات.

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

هذه طريقة تقليدية لكنها موثوقة للغاية للحصول على المقاييس. وفي الواقع، تعتمد واجهة المستخدم في Spark على الآلية نفسها لتصوّر المقاييس. تمكّن واجهة برمجة تطبيقات مستمعي Spark المطوّرين من تتبّع الأحداث التي يصدرها Spark أثناء تشغيل التطبيق. وتشمل هذه الأحداث عادةً: بدء التطبيق/انتهاؤه، وبدء المهمة/انتهاؤها، وبدء المرحلة/انتهاؤها، وغير ذلك. ويمكنك العثور على القائمة الكاملة لهذه الأحداث في توثيق Spark JavaDoc. ومن السهل تهيئة Spark Listeners واستخدامها لجمع المقاييس. بعد تنفيذ كل عملية، يقوم Spark باستدعاء Spark Listener ويمرّر إلى دوالّه مجموعة من البيانات الوصفية. وتتضمّن هذه البيانات معلومات مثل: زمن التنفيذ، وعدد السجلات المقروءة والمكتوبة، وحجم البيانات (بالبايت) المقروءة والمكتوبة، وغيرها.

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

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

ومع ذلك، يمكن أن يزوّدك Spark Listener البسيط برؤى مفيدة حول بياناتك.

فيما يلي مثال على Spark Listener من هذا النوع:

public class SomeSparkListener extends SparkListener {

    /**
     * This very simple spark listener prints metrics collected for every stage.
     *
     * @param stageCompleted
     */
    @Override
    public void onStageCompleted(SparkListenerStageCompleted stageCompleted) {
        StageInfo stageInfo = stageCompleted.stageInfo();
        Iterator it = stageInfo.taskMetrics().accumulators().iterator();

        while (it.hasNext()) {
            AccumulatorV2 next = it.next();
            String key = next.name().get();
            Object value = next.value();
            System.out.printf("key: %s, value: %s%n", key, value);
        }
    }
}

يمكنك إضافة Spark Listener إلى تطبيقك بعدة طرق:

إضافته برمجياً:

SparkSession spark = SparkSession.builder().getOrCreate();
spark.sparkContext().addSparkListener(new SomeSparkListener());

أو تمريره من خلال خيارات برنامج التشغيل في spark-submit أو في مشغّل ممجوعة Spark:

spark-submit --conf "spark.extraListeners=ai.databand.SomeSparkListener"

تمثّل هذه الآلية خياراً جاهزاً لمراقبة Spark. وبدل التركيز على المقاييس منخفضة المستوى جداً، يتيح Query Execution Listener للمطوّرين الاشتراك في أحداث إتمام الاستعلامات. ويقدّم بيانات وصفية أعلى مستوى عن الاستعلام المنفَّذ، مثل الخطة المنطقية، والخطة والفعلية، ومقاييس التنفيذ.

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

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

مع ذلك، قد تكون عملية استخراج البيانات من الخطة معقّدة لأن عليك التعامل مع واجهات برمجة تطبيقات Spark منخفضة المستوى.

إضافة إلى ذلك، تظل الأعباء التشغيلية الخاصة ببناء طبقة تخزين المقاييس وآليات التنبيه قائمة كما هي. فما تحصل عليه من Spark في النهاية هو بيانات وصفية فقط. وتقع مسؤولية استثمار هذه البيانات والاستفادة منها على عاتق المطوّر.

فيما يلي مثال على Query Execution Listener بسيط يقوم بطباعة الخطة والمقاييس:

public class ExampleQueryExecutionListener implements QueryExecutionListener {

    /**
     * Print plan and query metrics
     *
     * @param funcName
     * @param qe
     * @param durationNs
     */
    @Override
    public void onSuccess(String funcName, QueryExecution qe, long durationNs) {
        System.out.println(qe.executedPlan().prettyJson());
        Iterator it = qe.executedPlan().metrics().iterator();
        while (it.hasNext()) {
            Tuple2 next = it.next();
            System.out.printf("Key: %s, value: %s%n", next._1(), next._2().value());
        }
    }

    @Override
    public void onFailure(String funcName, QueryExecution qe, Exception exception) {
    }
}

يمكن إضافة Query execution listeners إما برمجيا أو عبر الإعدادات:

في كود التطبيق: SparkSession spark = SparkSession.builder().getOrCreate(); spark.listenerManager().register(new ExampleQueryExecutionListener());

عبر Spark-Relevance:

spark-submit --conf "spark.sql.queryExecutionListeners=ai.databand.ExampleQueryExecutionListener"

قد يكون تطبيق المراقبة منخفضة المستوى مهمة شاقة؛ غير أن أسلوب المراقبة على مستوى "النظام" يتمتع بميزة كبيرة: فهو لا يضيف أي عبء حسابي يُذكر. نظرًا لأن البيانات الوصفية تُولِّدها وتسجّلها الآليات الداخلية في Spark، فإن ذلك لا يفرض أي تأثيرات سلبية على أزمنة تنفيذ الاستعلامات.

يتيح لك استخدام Listeners للمراقبة تجنّب تعديل أي تعليمات برمجية للتطبيق. يمكن أن يحقق ذلك فوائد كبيرة عندما تريد تتبُّع البيانات في التطبيقات الحالية والتطبيقات القديمة (legacy) من دون توفّر ميزانية لإجراء التغييرات. كل ما عليك هو كتابة مستمع، وتمريره عبر إعدادات Spark لتحصل على صورة واضحة عن بياناتك.

يمكنك زيادة ثقتك بالبيانات الواردة بدرجة كبيرة من خلال التحقق من صحتها يدويًا. لنفترض أننا نتوقع عددًا معيّنًا من السجلات في مصدر بيانات الإدخال، وألا يقل هذا العدد عادةً عن X. يمكننا كتابة شيء بسيط للغاية مثل:

df = spark.read("path")
     if (df.count < X) {
     throw new RuntimeException("Input data is missing")
 }

الاحتمالات هنا غير محدودة. يمكننا مقارنة الأعداد، وعدد القيم غير الفارغة، والمخططات المستنتَجة، وغيرها.

نظرًا لأن العديد من فحوصات الجودة بسيطة نسبيًا، مثل التأكد من أن إطار البيانات لديك يتمتّع بالشكل والمحتويات المناسبة، فقد طوّر المجتمع مكتبات ملائمة لمثل هذه الفحوصات. إحدى هذه المكتبات هي Deequ. توفر Deequ لغة خاصة بالمجال (DSL) غنيّة تغطي معظم الحالات. جرِّبها. كما توفر خصائص متقدّمة، مثل توصيف الأعمدة، وحساب الحد الأدنى والحد الأقصى والمتوسط والنِّسَب المئوية، وحساب المدرّجات التكرارية (histograms)، والكشف عن الحالات الشاذة وغيرها الكثير.

تأمّل المثال التالي من وثائق Deequ:

val verificationResult = VerificationSuite()
   .onData(data)
   .addCheck(
     Check(CheckLevel.Error, "unit testing my data")
       .hasSize(_ == 5) // we expect 5 rows
       .isComplete("id") // should never be NULL
       .isUnique("id") // should not contain duplicates
       .isComplete("productName") // should never be NULL
       // should only contain the values "high" and "low"
       .isContainedIn("priority", Array("high", "low"))
       .isNonNegative("numViews") // should not contain negative values
       // at least half of the descriptions should contain a url
       .containsURL("description", _ >= 0.5)
       // half of the items should have less than 10 views
       .hasApproxQuantile("numViews", 0.5, _ <= 10))
     .run()

يمكن ملاحظة أن لدينا مجموعة كبيرة من اختبارات التحقق مُعرَّفة ضمن DSL جاهز للاستخدام الفوري.

والأهم أن Deequ يتيح لك تخزين نتائج هذه الاختبارات وإجراء مقارنات تلقائية مع نتائج التشغيلات السابقة. ويتم ذلك من خلال استخدام مستودعات المقاييس. ويمكنك تطوير تنفيذك الخاص لمستودع المقاييس ودمج Deequ بسلاسة ضمن البنية التحتية للمراقبة الموجودة في مؤسستك.

ورغم أن اختبارات جودة البيانات على مستوى التطبيق أكثر مرونة بكثير من الأساليب منخفضة المستوى، فإنها تأتي مع عيب واضح يتمثل في تأثيرها على الأداء. فكل عملية احتساب تُترجَم عمليًا إلى عملية إضافية في Spark، وقد يُنتج عن ذلك عبء تشغيلي واضح، خصوصًا مع مجموعات البيانات كبيرة الحجم. فحتى استعلامات بسيطة مثل "count" أو "where" يمكن أن تؤدي إلى مسح كامل لمجموعة البيانات. يبذل Spark ما في وسعه داخليًا لتحسين خطط التنفيذ، لكن يبقى من الضروري أخذ هذه الآثار في الحسبان والتأكد من أن "تنميط البيانات" لا ينعكس سلبًا على الأداء الكلي.

استعرضنا في هذا الدليل أكثر من نهج لمراقبة جودة البيانات في تطبيقات Spark. النهج منخفض المستوى يعتمد على واجهة برمجة تطبيقات Spark Event Listeners، ويمنحك الوصول إلى مقاييس تشغيل تفصيلية مثل عدد السجلات المقروءة/المكتوبة، والخطط المنطقية والفعلية، ما يسهّل بناء اتجاهات للمقاييس، والتحقق من أن مسار البيانات ينتج مخرجات سليمة، والحصول على رؤية عامة للتطبيقات القائمة من دون الحاجة إلى تعديل التعليمات البرمجية. أما الأساليب عالية المستوى، مثل فحص البيانات مباشرة أو استخدام مكتبات متخصصة في جودة البيانات، فهي أكثر سهولة في الاستخدام، لكنها غالبًا ما تأتي على حساب الأداء.

وكما هو معتاد في البيئات الواقعية، هناك دائمًا مفاضلات، ولكل نهج سيناريوهات يكون فيها الأنسب، وفقًا لطبيعة التطبيق ومتطلبات الأعمال. وعليك الموازنة بين هذه الاستخدامات.

في IBM® Databand نعتمد النهجين معًا لنوفّر مجموعة متكاملة من الخيارات لتتبّع تطبيقات Spark ومراقبتها. فعلى مستوى الأساس نستخدم Spark Listeners لبناء اتجاهات المقاييس وتتبع دورة حياة البيانات، كما نوفر “مخزن مقاييس” مدمجًا مع Deequ، إلى جانب إمكان تتبّع المقاييس التي تُحتسب يدويًا على مستوى كل مقياس على حدة.

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