معالجة الدفق باستخدام Azure Stream Analytics

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

يتوفر تنفيذ مرجعي لهذه البنية على GitHub.

بناء الأنظمة

قم بتنزيل ملف Visio لهذه البنية.

‏‏سير العمل‬

تتكون البنية من المكونات التالية:

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

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

Azure Stream Analytics. Stream Analytics هو محرك لمعالجة الأحداث. تقرأ وظيفة Stream Analytics تدفقات البيانات من مركزي الأحداث وتنفذ معالجة التدفق.

قاعدة بيانات Azure Cosmos. الإخراج من وظيفة Stream Analytics هو سلسلة من السجلات، والتي تتم كتابتها كمستندات JSON إلى قاعدة بيانات مستندات Azure Cosmos DB.

Microsoft Power BI. إن Power BI عبارة عن مجموعة من أدوات تحليل الأعمال لتحليل البيانات من أجل رؤى الأعمال. في هذه البنية، يقوم بتحميل البيانات من Azure Cosmos DB. يتيح ذلك للمستخدمين تحليل المجموعة الكاملة من البيانات التاريخية التي تم جمعها. يمكنك أيضاً دفق النتائج مباشرةً من Stream Analytics إلى Power BI لعرض البيانات في الوقت الحقيقي. لمزيد من المعلومات، راجع التدفق في الوقت الحقيقي في Power BI.

Azure Monitor. يجمع Azure Monitor قياسات الأداء بشأن خدمات Azure المنتشرة في الحل. من خلال تصور هذه في لوحة القيادة، يمكنك الحصول على رؤى بشأن صحة الحل.

تفاصيل السيناريو

السيناريو: تجمع شركة سيارات الأجرة بيانات حول كل رحلة تاكسي. بالنسبة لهذا السيناريو، نفترض وجود جهازين منفصلين يرسلان البيانات. يوجد في سيارة الأجرة عداد يرسل معلومات بشأن كل رحلة - المدة، والمسافة، ومواقع الالتقاء والتوصيل. يقبل جهاز منفصل المدفوعات من العملاء ويرسل بيانات حول الأسعار. تريد شركة سيارات الأجرة حساب متوسط ​​الإكرامية لكل ميل يتم قيادته، في الوقت الحقيقي، من أجل تحديد الاتجاهات.

حالات الاستخدام المحتملة

تم تحسين هذا الحل لسيناريو البيع بالتجزئة.

احتواء البيانات

لمحاكاة مصدر بيانات، تستخدم هذه البنية المرجعية مجموعة بيانات سيارات الأجرة في مدينة نيويورك رقم ^[1]. تحتوي مجموعة البيانات هذه على بيانات بشأن رحلات سيارات الأجرة في مدينة نيويورك على مدار أربع سنوات (2010-2013). يحتوي على نوعين من السجلات: بيانات الركوب وبيانات الأجرة. تتضمن بيانات الرحلة مدة الرحلة، ومسافة الرحلة، وموقع الالتقاء والتوصيل. تتضمن بيانات الأجرة مبالغ الأجرة والضرائب والإكرامية. تتضمن الحقول الشائعة في كلا نوعي السجلات رقم لوحة السيارة ورخصة القيادة وبطاقة هوية المورد. تحدد هذه الحقول الثلاثة معاً بشكل فريد سيارة الأجرة بالإضافة إلى السائق. يتم تخزين البيانات بتنسيق CSV.

[1] دونوفان، بريان؛ العمل، دان (2016): بيانات رحلة تاكسي مدينة نيويورك (2010-2013). جامعة إلينوي في أوربانا شامبين. https://doi.org/10.13012/J8PN93H8

منشئ البيانات هو تطبيق ‎.NET Core يقرأ السجلات ويرسلها إلى مراكز الأحداث. يرسل المنشئ بيانات الركوب بتنسيق JSON وبيانات الأجرة بتنسيق CSV.

تستخدم مراكز الأحداث أقسامًا لتقسيم البيانات. تسمح الأقسام للمستهلك بقراءة كل قسم على التوازي. عند إرسال البيانات إلى مراكز الأحداث (مراكز الأحداث)، يمكنك تحديد مفتاح القسم بشكل صريح. وبخلاف ذلك، يتم تعيين السجلات إلى أقسام بأسلوب round-robin.

في هذا السيناريو المحدد، يجب أن تنتهي بيانات الركوب وبيانات الأجرة بنفس معرف القسم لكابينة تاكسي معينة. يمكّن هذا Stream Analytics من تطبيق درجة من التوازي عندما يربط بين الدفقين. سيتطابق السجل الموجود في القسم و من بيانات الرحلة مع السجل الموجود في القسم و بيانات الأجرة.

في منشئ البيانات، يحتوي نموذج البيانات المشترك لكلا نوعي السجلات على خاصية PartitionKey وهي عبارة عن تسلسل Medallion وHackLicense وVendorId.

public abstract class TaxiData
{
    public TaxiData()
    {
    }

    [JsonProperty]
    public long Medallion { get; set; }

    [JsonProperty]
    public long HackLicense { get; set; }

    [JsonProperty]
    public string VendorId { get; set; }

    [JsonProperty]
    public DateTimeOffset PickupTime { get; set; }

    [JsonIgnore]
    public string PartitionKey
    {
        get => $"{Medallion}_{HackLicense}_{VendorId}";
    }

تُستخدم هذه الخاصية لتوفير مفتاح قسم صريح عند الإرسال إلى مراكز الأحداث:

using (var client = pool.GetObject())
{
    return client.Value.SendAsync(new EventData(Encoding.UTF8.GetBytes(
        t.GetData(dataFormat))), t.PartitionKey);
}

المعالجة المتدفقة

يتم تحديد مهمة معالجة الدفق باستخدام استعلام SQL مع عدة خطوات مميزة. تقوم الخطوتان الأوليان بتحديد السجلات من دفقَي الإدخال.

WITH
Step1 AS (
    SELECT PartitionId,
           TRY_CAST(Medallion AS nvarchar(max)) AS Medallion,
           TRY_CAST(HackLicense AS nvarchar(max)) AS HackLicense,
           VendorId,
           TRY_CAST(PickupTime AS datetime) AS PickupTime,
           TripDistanceInMiles
    FROM [TaxiRide] PARTITION BY PartitionId
),
Step2 AS (
    SELECT PartitionId,
           medallion AS Medallion,
           hack_license AS HackLicense,
           vendor_id AS VendorId,
           TRY_CAST(pickup_datetime AS datetime) AS PickupTime,
           tip_amount AS TipAmount
    FROM [TaxiFare] PARTITION BY PartitionId
),

تنضم الخطوة التالية إلى دفقَي الإدخال لتحديد السجلات المطابقة من كل تدفق.

Step3 AS (
  SELECT tr.TripDistanceInMiles,
         tf.TipAmount
    FROM [Step1] tr
    PARTITION BY PartitionId
    JOIN [Step2] tf PARTITION BY PartitionId
      ON tr.PartitionId = tf.PartitionId
     AND tr.PickupTime = tf.PickupTime
     AND DATEDIFF(minute, tr, tf) BETWEEN 0 AND 15
)

يقوم هذا الاستعلام بضم السجلات في مجموعة من الحقول التي تحدد بشكل فريد السجلات المتطابقة (PartitionId وPickupTime).

في Stream Analytics، تكون الصلات مؤقتة، ما يعني أن السجلات يتم ضمها خلال فترة زمنية معينة. خلاف ذلك، قد تحتاج الوظيفة إلى الانتظار إلى أجل غير مسمى للمباراة. تحدد الوظيفة DATEDIFF مدى إمكانية فصل سجلين متطابقين في الوقت المناسب للمباراة.

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

SELECT System.Timestamp AS WindowTime,
       SUM(tr.TipAmount) / SUM(tr.TripDistanceInMiles) AS AverageTipPerMile
  INTO [TaxiDrain]
  FROM [Step3] tr
  GROUP BY HoppingWindow(Duration(minute, 5), Hop(minute, 1))

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

في البنية الموضحة هنا، يتم حفظ نتائج مهمة Stream Analytics فقط في Azure Cosmos DB. بالنسبة لسيناريو البيانات الضخمة، ضع في اعتبارك أيضاً استخدام مراكز الأحداث Capture لحفظ بيانات الأحداث الأولية في تخزين Azure Blob. سيسمح لك الاحتفاظ بالبيانات الأولية بتشغيل استعلامات مجمعة على بياناتك التاريخية في وقت لاحق، من أجل استخلاص رؤى جديدة من البيانات.

الاعتبارات

تنفذ هذه الاعتبارات ركائز Azure Well-Architected Framework، وهو عبارة عن مجموعة من المبادئ التوجيهية التي يمكن استخدامها لتحسين جودة حمل العمل. لمزيد من المعلومات، يرجى مراجعةMicrosoft Azure Well-Architected Framework.

قابلية التوسع

مراكز الأحداث

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

Stream Analytics

بالنسبة لـ Stream Analytics، يتم قياس موارد الحوسبة المخصصة لوظيفة ما في وحدات التدفق. مقياس وظائف Stream Analytics أفضل إذا كان من الممكن موازاة الوظيفة. بهذه الطريقة، يمكن لـ Stream Analytics توزيع المهمة عبر عقد حوسبة متعددة.

بالنسبة لإدخال مراكز الأحداث، استخدم الكلمة الرئيسية PARTITION BY لتقسيم وظيفة Stream Analytics. سيتم تقسيم البيانات إلى مجموعات فرعية بناءً على أقسام مراكز الأحداث.

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

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

• الخطوتان 1 و2 عبارة عن عبارات SELECT بسيطة تحدد السجلات داخل قسم واحد.
• تقوم الخطوة 3 بتنفيذ صلة مقسمة عبر دفقين للإدخال. تستفيد هذه الخطوة من حقيقة أن السجلات المطابقة تشترك في نفس مفتاح القسم، وبالتالي فهي مضمونة أن يكون لها نفس معرف القسم في كل دفق إدخال.
• يتم تجميع الخطوة 4 عبر جميع الأقسام. هذه الخطوة لا يمكن أن تكون متوازية.

استخدم Stream Analytics مخطط الوظيفة لمعرفة عدد الأقسام التي تم تعيينها لكل خطوة في الوظيفة. يوضح الرسم البياني التالي مخطط الوظيفة لهذه البنية المرجعية:

Azure Cosmos DB

تقاس سعة معدل النقل ل Azure Cosmos DB بوحدات الطلب (RUs). من أجل تغيير حجم حاوية Azure Cosmos DB بعد 10000 وحدة طلب، يجب تحديد مفتاح قسم عند إنشاء الحاوية، وتضمين مفتاح القسم في كل مستند.

في هذه البنية المرجعية، يتم إنشاء المستندات الجديدة مرة واحدة فقط في الدقيقة (الفاصل الزمني لإطار التنقل)، وبالتالي فإن متطلبات معدل النقل منخفضة جداً. لهذا السبب، ليست هناك حاجة لتعيين مفتاح قسم في هذا السيناريو.

مراقبة‬

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

تشير أي من إشارات التحذير التالية إلى أنه يجب عليك توسيع نطاق مورد Azure ذي الصلة:

• تعمل مراكز الأحداث على تقليص الطلبات أو الاقتراب من حصة الرسائل اليومية.
• تستخدم وظيفة Stream Analytics باستمرار أكثر من 80% من وحدات البث المخصصة (SU).
• يبدأ Azure Cosmos DB في تقييد الطلبات.

تتضمن البنية المرجعية لوحة معلومات مخصصة، يتم توزيعها في مدخل Microsoft Azure. بعد توزيع البنية، يمكنك عرض لوحة المعلومات عن طريق فتح مدخل Microsoft Azure وتحديد TaxiRidesDashboard من قائمة لوحات المعلومات. لمزيد من المعلومات بشأن إنشاء وتوزيع لوحات معلومات مخصصة في مدخل Microsoft Azure، راجع إنشاء لوحات معلومات Azure برمجياً.

تُظهر الصورة التالية لوحة القيادة بعد تشغيل وظيفة Stream Analytics لمدة ساعة تقريباً.

تُظهر اللوحة الموجودة في أسفل اليسار أن استهلاك وحدة التخزين لوظيفة Stream Analytics يرتفع خلال أول 15 دقيقة ثم يتوقف عن العمل. هذا نمط نموذجي حيث تصل الوظيفة إلى حالة مستقرة.

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

تم تمكين النفخ التلقائي في حوالي الساعة 06:35. يمكنك رؤية انخفاض p في الطلبات المخنوقة، حيث قامت مراكز الأحداث تلقائياً بتوسيع نطاق يصل إلى 3 وحدات معدل نقل.

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

تحسين التكلفة

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

استخدم حاسبة تسعير Azure لتقدير التكاليف. فيما يلي بعض الاعتبارات للخدمات المستخدمة في هذه البنية المرجعية.

Azure Stream Analytics

يتم أسعار Azure Stream Analytics بعدد وحدات التدفق (0.11 دولار/ساعة) المطلوبة لمعالجة البيانات في الخدمة.

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

مراكز الأحداث وAzure Cosmos DB

للحصول على اعتبارات التكلفة حول Azure Event Hubs وAzure Cosmos DB، راجع اعتبارات التكلفة راجع معالجة الدفق باستخدام البنية المرجعية ل Azure Databricks .

DevOps

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

• استخدم قالب Azure Resource Manager لتوزيع موارد Azure باتباع عملية البنية الأساسية كتعليمة برمجية (IaC). باستخدام القوالب، تكون أتمتة عمليات التوزيع باستخدام خدمات Azure DevOpsأو حلول CI / CD الأخرى أسهل.

• ضع كل حمل عمل في قالب توزيع منفصل، وقم بتخزين الموارد في أنظمة التحكم بالمصادر. يمكنك توزيع القوالب معاً أو بشكل فردي كجزء من عملية CI/CD، مما يجعل عملية الأتمتة أسهل.

  في هذه البنية، يتم تحديد Azure Event Hubs وLog Analytics وAzure Cosmos DB على أنها حمل عمل واحد. يتم تضمين هذه الموارد في قالب ARM واحد.

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

• ضع في اعتبارك استخدام Azure Monitor لتحليل أداء مسار معالجة التدفق. لمزيد من المعلومات، راجع مراقبة Azure Databricks.

لمزيد من المعلومات، راجع ركيزة التميز التشغيلي في Microsoft Azure Well-Architected Framework.

نشر هذا السيناريو

لنشر وتشغيل التطبيق المرجعي، اتبع الخطوات في الملف التمهيدي GitHub.

الموارد ذات الصلة

قد تحتاج إلى مراجعة سيناريو مثال Azure التالي الذي يوضح حلا معينا باستخدام بعض التقنيات نفسها:

• الكشف عن الاحتيال في الوقت الحقيقي