Tarif: Azure AI hizmetleri - Çok Değişkenli Anomali Algılama
Bu tarif, çok değişkenli anomali algılama için Apache Spark üzerinde SynapseML ve Azure AI hizmetlerini nasıl kullanabileceğinizi gösterir. Çok değişkenli anomali algılama, farklı değişkenler arasındaki tüm bağıntıları ve bağımlılıkları dikkate alarak birçok değişken veya zaman serisi arasındaki anomalilerin algılanmasına olanak tanır. Bu senaryoda, Azure AI hizmetlerini kullanarak bir modeli çok değişkenli anomali algılama için eğitmek için SynapseML'yi kullanırız ve ardından modele kullanarak üç IoT algılayıcısından yapay ölçümler içeren bir veri kümesi içindeki çok değişkenli anomalileri çıkaracağız.
Önemli
20 Eylül 2023 tarihinden itibaren yeni Anomali Algılayıcısı kaynakları oluşturamayacaksınız. Anomali Algılayıcısı hizmeti 1 Ekim 2026'da kullanımdan kaldırılıyor.
Azure AI Anomali Algılayıcısı hakkında daha fazla bilgi edinmek için bu belgeler sayfasına bakın.
Önkoşullar
- Azure aboneliği - Ücretsiz bir abonelik oluşturun
- Not defterinizi bir göle ekleyin. Sol tarafta Ekle'yi seçerek mevcut bir göl evi ekleyin veya bir göl evi oluşturun.
Ayarlama
Azure portalını kullanarak kaynak Anomaly Detector oluşturmak için yönergeleri izleyin veya alternatif olarak bu kaynağı oluşturmak için Azure CLI'yı da kullanabilirsiniz.
bir Anomaly Detectorayarladıktan sonra, çeşitli biçimlerdeki verileri işleme yöntemlerini keşfedebilirsiniz. Azure AI içindeki hizmet kataloğu çeşitli seçenekler sunar: Görüntü İşleme, Konuşma, Dil, Web araması, Karar, Çeviri ve Belge Zekası.
Anomali Algılayıcısı kaynağı oluşturma
- Azure portalında kaynak grubunuzda Oluştur'u seçin ve Anomali Algılayıcısı yazın. Anomali Algılayıcısı kaynağını seçin.
- Kaynağa bir ad verin ve ideal olarak kaynak grubunuzun geri kalanıyla aynı bölgeyi kullanın. Geri kalanı için varsayılan seçenekleri kullanın ve gözden geçir + Oluştur'u ve ardından Oluştur'u seçin.
- Anomali Algılayıcısı kaynağı oluşturulduktan sonra açın ve sol gezinti bölmesindeki paneli seçin
Keys and Endpoints. Anomali Algılayıcısı kaynağının anahtarını ortam değişkenineANOMALY_API_KEYkopyalayın veya değişkendeanomalyKeydepolayın.
Depolama Hesabı kaynağı oluşturma
Ara verileri kaydetmek için bir Azure Blob Depolama Hesabı oluşturmanız gerekir. Bu depolama hesabında ara verileri depolamak için bir kapsayıcı oluşturun. Kapsayıcı adını not edin ve bağlantı dizesi bu kapsayıcıya kopyalayın. Değişkeni ve BLOB_CONNECTION_STRING ortam değişkenini containerName doldurmak için daha sonra gerekir.
Hizmet anahtarlarınızı girin
Hizmet anahtarlarımız için ortam değişkenlerini ayarlayarak başlayalım. Sonraki hücre, Ve BLOB_CONNECTION_STRING ortam değişkenlerini Azure Key Vault'umuzda depolanan değerlere göre ayarlarANOMALY_API_KEY. Bu öğreticiyi kendi ortamınızda çalıştırıyorsanız, devam etmeden önce bu ortam değişkenlerini ayarladığınızdan emin olun.
import os
from pyspark.sql import SparkSession
from synapse.ml.core.platform import find_secret
# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()
Şimdi ve BLOB_CONNECTION_STRING ortam değişkenlerini okuyup ANOMALY_API_KEY ve location değişkenlerini ayarlayalımcontainerName.
# An Anomaly Dectector subscription key
anomalyKey = find_secret("anomaly-api-key") # use your own anomaly api key
# Your storage account name
storageName = "anomalydetectiontest" # use your own storage account name
# A connection string to your blob storage account
storageKey = find_secret("madtest-storage-key") # use your own storage key
# A place to save intermediate MVAD results
intermediateSaveDir = (
"wasbs://madtest@anomalydetectiontest.blob.core.windows.net/intermediateData"
)
# The location of the anomaly detector resource that you created
location = "westus2"
Anomali algılayıcının ara sonuçları kaydedebilmesi için öncelikle depolama hesabımıza bağlanıyoruz:
spark.sparkContext._jsc.hadoopConfiguration().set(
f"fs.azure.account.key.{storageName}.blob.core.windows.net", storageKey
)
Şimdi tüm gerekli modülleri içeri aktaralım.
import numpy as np
import pandas as pd
import pyspark
from pyspark.sql.functions import col
from pyspark.sql.functions import lit
from pyspark.sql.types import DoubleType
import matplotlib.pyplot as plt
import synapse.ml
from synapse.ml.cognitive import *
Şimdi örnek verilerimizi bir Spark DataFrame'de okuyalım.
df = (
spark.read.format("csv")
.option("header", "true")
.load("wasbs://publicwasb@mmlspark.blob.core.windows.net/MVAD/sample.csv")
)
df = (
df.withColumn("sensor_1", col("sensor_1").cast(DoubleType()))
.withColumn("sensor_2", col("sensor_2").cast(DoubleType()))
.withColumn("sensor_3", col("sensor_3").cast(DoubleType()))
)
# Let's inspect the dataframe:
df.show(5)
Artık modelimizi eğitmek için kullanılan bir estimator nesne oluşturabiliriz. Eğitim verileri için başlangıç ve bitiş saatlerini belirtiriz. Ayrıca kullanılacak giriş sütunlarını ve zaman damgalarını içeren sütunun adını belirtiriz. Son olarak, anomali algılama kayan penceresinde kullanılacak veri noktalarının sayısını belirtir ve bağlantı dizesi Azure Blob Depolama Hesabı olarak ayarlarız.
trainingStartTime = "2020-06-01T12:00:00Z"
trainingEndTime = "2020-07-02T17:55:00Z"
timestampColumn = "timestamp"
inputColumns = ["sensor_1", "sensor_2", "sensor_3"]
estimator = (
FitMultivariateAnomaly()
.setSubscriptionKey(anomalyKey)
.setLocation(location)
.setStartTime(trainingStartTime)
.setEndTime(trainingEndTime)
.setIntermediateSaveDir(intermediateSaveDir)
.setTimestampCol(timestampColumn)
.setInputCols(inputColumns)
.setSlidingWindow(200)
)
öğesini oluşturduğumuza estimatorgöre verilerine sığdıralım:
model = estimator.fit(df)
```parameter
Once the training is done, we can now use the model for inference. The code in the next cell specifies the start and end times for the data we would like to detect the anomalies in.
```python
inferenceStartTime = "2020-07-02T18:00:00Z"
inferenceEndTime = "2020-07-06T05:15:00Z"
result = (
model.setStartTime(inferenceStartTime)
.setEndTime(inferenceEndTime)
.setOutputCol("results")
.setErrorCol("errors")
.setInputCols(inputColumns)
.setTimestampCol(timestampColumn)
.transform(df)
)
result.show(5)
Önceki hücreyi çağırdığımızda .show(5) , veri çerçevesinin ilk beş satırını gösterdi. Sonuçların tümü null çıkarım penceresinin içinde olmadıkları için elde edildi.
Yalnızca çıkarsanan verilerin sonuçlarını göstermek için ihtiyacımız olan sütunları seçmeliyiz. Ardından veri çerçevesindeki satırları artan düzende sıralayabilir ve sonucu yalnızca çıkarım penceresinin aralığındaki satırları gösterecek şekilde filtreleyebiliriz. Bizim örneğimizde inferenceEndTime veri çerçevesinin son satırıyla aynıdır, bu nedenle bunu yoksayabilirsiniz.
Son olarak, sonuçları daha iyi çizebilmek için Spark veri çerçevesini Pandas veri çerçevesine dönüştürün.
rdf = (
result.select(
"timestamp",
*inputColumns,
"results.contributors",
"results.isAnomaly",
"results.severity"
)
.orderBy("timestamp", ascending=True)
.filter(col("timestamp") >= lit(inferenceStartTime))
.toPandas()
)
rdf
Her algılayıcıdan contributors algılanan anomalilere katkı puanını depolayan sütunu biçimlendirin. Sonraki hücre bu verileri biçimlendirip her algılayıcının katkı puanını kendi sütununa böler.
def parse(x):
if type(x) is list:
return dict([item[::-1] for item in x])
else:
return {"series_0": 0, "series_1": 0, "series_2": 0}
rdf["contributors"] = rdf["contributors"].apply(parse)
rdf = pd.concat(
[rdf.drop(["contributors"], axis=1), pd.json_normalize(rdf["contributors"])], axis=1
)
rdf
Harika! Artık sırasıyla , series_1ve series_2 sütunlarında 1, 2 ve 3 algılayıcılarının series_0katkı puanlarına sahibiz.
Sonuçları çizmek için sonraki hücreyi çalıştırın. minSeverity parametresi, çizilecek anomalilerin en düşük önem derecesini belirtir.
minSeverity = 0.1
####### Main Figure #######
plt.figure(figsize=(23, 8))
plt.plot(
rdf["timestamp"],
rdf["sensor_1"],
color="tab:orange",
linestyle="solid",
linewidth=2,
label="sensor_1",
)
plt.plot(
rdf["timestamp"],
rdf["sensor_2"],
color="tab:green",
linestyle="solid",
linewidth=2,
label="sensor_2",
)
plt.plot(
rdf["timestamp"],
rdf["sensor_3"],
color="tab:blue",
linestyle="solid",
linewidth=2,
label="sensor_3",
)
plt.grid(axis="y")
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.legend()
anoms = list(rdf["severity"] >= minSeverity)
_, _, ymin, ymax = plt.axis()
plt.vlines(np.where(anoms), ymin=ymin, ymax=ymax, color="r", alpha=0.8)
plt.legend()
plt.title(
"A plot of the values from the three sensors with the detected anomalies highlighted in red."
)
plt.show()
####### Severity Figure #######
plt.figure(figsize=(23, 1))
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.plot(
rdf["timestamp"],
rdf["severity"],
color="black",
linestyle="solid",
linewidth=2,
label="Severity score",
)
plt.plot(
rdf["timestamp"],
[minSeverity] * len(rdf["severity"]),
color="red",
linestyle="dotted",
linewidth=1,
label="minSeverity",
)
plt.grid(axis="y")
plt.legend()
plt.ylim([0, 1])
plt.title("Severity of the detected anomalies")
plt.show()
####### Contributors Figure #######
plt.figure(figsize=(23, 1))
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.bar(
rdf["timestamp"], rdf["series_0"], width=2, color="tab:orange", label="sensor_1"
)
plt.bar(
rdf["timestamp"],
rdf["series_1"],
width=2,
color="tab:green",
label="sensor_2",
bottom=rdf["series_0"],
)
plt.bar(
rdf["timestamp"],
rdf["series_2"],
width=2,
color="tab:blue",
label="sensor_3",
bottom=rdf["series_0"] + rdf["series_1"],
)
plt.grid(axis="y")
plt.legend()
plt.ylim([0, 1])
plt.title("The contribution of each sensor to the detected anomaly")
plt.show()
Çizimler, algılayıcılardan alınan ham verileri (çıkarım penceresinin içinde) turuncu, yeşil ve mavi olarak gösterir. İlk şekildeki kırmızı dikey çizgiler, önem derecesi değerinden büyük veya eşit minSeverityolan algılanan anomalileri gösterir.
İkinci çizimde, algılanan tüm anomalilerin önem derecesi puanı gösterilir ve minSeverity eşik noktalı kırmızı çizgide gösterilir.
Son olarak, son çizimde her algılayıcıdan alınan verilerin algılanan anomalilere katkısı gösterilir. Her anomalinin en olası nedenini tanılamamıza ve anlamamıza yardımcı olur.