وصفة: الصيانة التنبؤية مع الخدمات المعرفية للبيانات الضخمة

  • مقالة
  • قراءة خلال 2 دقائق

توضح هذه الوصفة كيف يمكنك استخدام Azure Synapse Analytics والخدمات المعرفية على Apache Spark للصيانة التنبؤية لأجهزة إنترنت الأشياء. سنتابع مع عينة CosmosDB و Synapse Link . للحفاظ على بساطة الأمور ، في هذه الوصفة ، سنقرأ البيانات مباشرة من ملف CSV بدلا من الحصول على البيانات المتدفقة من خلال CosmosDB و Synapse Link. نحن نشجعك بشدة على إلقاء نظرة على عينة Synapse Link.

سيناريو افتراضي

السيناريو الافتراضي هو محطة توليد الكهرباء ، حيث تقوم أجهزة إنترنت الأشياء بمراقبة التوربينات البخارية. تحتوي مجموعة IoTSignals على بيانات ثورات في الدقيقة (RPM) وميجاوات (MW) لكل توربين. يتم تحليل الإشارات من التوربينات البخارية ويتم الكشف عن الإشارات الشاذة.

يمكن أن يكون هناك قيم شاذة في البيانات بتردد عشوائي. في هذه الحالات ، سترتفع قيم RPM وسينخفض إنتاج MW ، لحماية الدائرة. الفكرة هي رؤية البيانات تختلف في نفس الوقت ، ولكن مع إشارات مختلفة.

المتطلبات الأساسية

الإعداد

إنشاء مورد Anomaly Detector

يتم تمثيل Azure Cognitive Services بواسطة موارد Azure التي تشترك فيها. قم بإنشاء مورد المترجم باستخدام مدخل Azure أو Azure CLI. يمكنك أيضًا:

دون نقطة النهاية ومفتاح هذا المورد ، ستحتاج إليه في هذا الدليل.

أدخل مفاتيح الخدمة

لنبدأ بإضافة مفتاحك وموقعك.

service_key = None # Paste your anomaly detector key here
location = None # Paste your anomaly detector location here

assert (service_key is not None)
assert (location is not None)

قراءة البيانات في إطار بيانات

بعد ذلك ، دعنا نقرأ ملف IoTSignals في DataFrame. افتح دفتر ملاحظات جديدا في مساحة عمل Synapse وقم بإنشاء DataFrame من الملف.

df_signals = spark.read.csv("wasbs://publicwasb@mmlspark.blob.core.windows.net/iot/IoTSignals.csv", header=True, inferSchema=True)

تشغيل الكشف عن الشذوذ باستخدام الخدمات المعرفية على Spark

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

from pyspark.sql.functions import col, struct
from mmlspark.cognitive import SimpleDetectAnomalies
from mmlspark.core.spark import FluentAPI

detector = (SimpleDetectAnomalies()
    .setSubscriptionKey(service_key)
    .setLocation(location)
    .setOutputCol("anomalies")
    .setGroupbyCol("grouping")
    .setSensitivity(95)
    .setGranularity("secondly"))

df_anomaly = (df_signals
    .where(col("unitSymbol") == 'RPM')
    .withColumn("timestamp", col("dateTime").cast("string"))
    .withColumn("value", col("measureValue").cast("double"))
    .withColumn("grouping", struct("deviceId"))
    .mlTransform(detector)).cache()

df_anomaly.createOrReplaceTempView('df_anomaly')

دعونا نلقي نظرة على البيانات:

df_anomaly.select("timestamp","value","deviceId","anomalies.isAnomaly").show(3)

يجب أن تسفر هذه الخلية عن نتيجة تبدو كما يلي:

طابع زمني القيمة معرف الجهاز isAnomaly
2020-05-01 18:33:51 3174 ديف-7 خطأ
2020-05-01 18:33:52 2976 ديف-7 خطأ
2020-05-01 18:33:53 2714 ديف-7 خطأ

تصور الحالات الشاذة لأحد الأجهزة

يحتوي IoTSignals.csv على إشارات من أجهزة إنترنت الأشياء المتعددة. سنركز على جهاز معين وتصور المخرجات الشاذة من الجهاز.

df_anomaly_single_device = spark.sql("""
select
  timestamp,
  measureValue,
  anomalies.expectedValue,
  anomalies.expectedValue + anomalies.upperMargin as expectedUpperValue,
  anomalies.expectedValue - anomalies.lowerMargin as expectedLowerValue,
  case when anomalies.isAnomaly=true then 1 else 0 end as isAnomaly
from
  df_anomaly
where deviceid = 'dev-1' and timestamp < '2020-04-29'
order by timestamp
limit 200""")

الآن بعد أن أنشأنا إطار بيانات يمثل الحالات الشاذة لجهاز معين ، يمكننا تصور هذه الحالات الشاذة:

import matplotlib.pyplot as plt
from pyspark.sql.functions import col

adf = df_anomaly_single_device.toPandas()
adf_subset = df_anomaly_single_device.where(col("isAnomaly") == 1).toPandas()

plt.figure(figsize=(23,8))
plt.plot(adf['timestamp'],adf['expectedUpperValue'], color='darkred', linestyle='solid', linewidth=0.25, label='UpperMargin')
plt.plot(adf['timestamp'],adf['expectedValue'], color='darkgreen', linestyle='solid', linewidth=2, label='Expected Value')
plt.plot(adf['timestamp'],adf['measureValue'], 'b', color='royalblue', linestyle='dotted', linewidth=2, label='Actual')
plt.plot(adf['timestamp'],adf['expectedLowerValue'],  color='black', linestyle='solid', linewidth=0.25, label='Lower Margin')
plt.plot(adf_subset['timestamp'],adf_subset['measureValue'], 'ro', label = 'Anomaly')
plt.legend()
plt.title('RPM Anomalies with Confidence Intervals')
plt.show()

إذا نجحت ، فسيبدو الإخراج الخاص بك كما يلي:

Anomaly Detector Plot

الخطوات التالية

تعرف على كيفية إجراء الصيانة التنبؤية على نطاق واسع باستخدام Azure Cognitive Services وAzure Synapse Analytics وAzure CosmosDB. لمزيد من المعلومات، راجع النموذج الكامل على GitHub.