Azure Stream Analytics ile akış işleme

Bu başvuru mimarisinde, uçtan uca akış işleme işlem hattı gösterilmektedir. Ardışık düzen, iki kaynaktan gelen verileri alır, iki akıştaki kayıtları ilişkilendirir ve bir zaman penceresi boyunca bir ortalama toplama değeri hesaplar. Sonuçlar daha fazla analiz için depolanır.

bu mimari için bir başvuru uygulamasına GitHubkullanılabilir.

Senaryo: bir TAXI şirketi her bir vergi yolculuğuna ilişkin veri toplar. Bu senaryo için, veri gönderen iki ayrı cihazın olduğunu varsaytık. Taxı her bir arttırıldığında hakkında bilgi gönderen bir ölçerdir: süre, uzaklık ve toplama ve açılan konumlar. Ayrı bir cihaz, müşterilerden gelen ödemeleri kabul eder ve Fares hakkındaki verileri gönderir. Taksi şirketi, eğilimleri belirlemek için gerçek zamanlı olarak mil başına ortalama ipucunu hesaplamak istiyor.

Mimari

Mimari aşağıdaki bileşenlerden oluşur.

Veri kaynakları. Bu mimaride, gerçek zamanlı veri akışları üreten iki veri kaynağı vardır. İlk akış, arttırıldığında bilgilerini içerir ve ikincisi tarifeli havayolu bilgilerini içerir. Başvuru mimarisi, bir statik dosya kümesinden okuyan ve verileri Event Hubs veren bir sanal veri Oluşturucu içerir. Gerçek bir uygulamada veri kaynakları, TAXI Cabs 'de yüklü cihazlar olacaktır.

Azure Event Hubs. Event Hubs bir olay alma hizmetidir. Bu mimari, her veri kaynağı için bir tane olmak üzere iki olay hub örneği kullanır. Her veri kaynağı, ilişkili olay hub 'ına bir veri akışı gönderir.

Azure Stream Analytics. Stream Analytics bir olay işleme motorudur. Stream Analytics işi, iki olay hub 'larından veri akışlarını okur ve akış işlemeyi gerçekleştirir.

Cosmos DB. Stream Analytics işi çıktısı, bir Cosmos DB belge veritabanına JSON belgeleri olarak yazılmış bir dizi kayıt olur.

Microsoft Power BI. Power BI, iş öngörülerine yönelik verileri çözümlemek için bir iş analizi araçları paketidir. bu mimaride, Cosmos DB verileri yükler. Bu, kullanıcıların toplanan geçmiş verilerin tamamını analiz etmesine olanak tanır. ayrıca, verilerin gerçek zamanlı bir görünümü için, sonuçları doğrudan Stream Analytics Power BI olarak akışla aktarabilirsiniz. Daha fazla bilgi için bkz. Power BI’da gerçek zamanlı akış yapma.

Azure İzleyici. Azure izleyici , çözümde dağıtılan Azure hizmetleriyle ilgili performans ölçümlerini toplar. Bunları bir panoda görselleştirerek çözümün sistem durumu hakkında öngörüler elde edebilirsiniz.

Veri alımı

Bir veri kaynağının benzetimini yapmak için, bu başvuru mimarisi New York City TAXI veri veri kümesini^[1]kullanır. Bu veri kümesi, New York şehrindeki vergilenme hakkındaki verileri dört yıllık bir dönemde (2010 – 2013) içerir. İki tür kayıt içerir: arttırıldığında Data ve tarifeli havayolu Data. Veri verileri, seyahat süresini, seyahat mesafesini ve toplama ve Dropoff konumunu içerir. Tarifeli havayolu verileri tarifeli havayolu, vergi ve tıp tutarlarını içerir. Her iki kayıt türündeki ortak alanlar arasında medalon numarası, Hack lisansı ve satıcı KIMLIĞI dahildir. Bu üç alan birlikte bir TAXI ve bir sürücüyü benzersiz şekilde tanımlar. Veriler CSV biçiminde depolanır.

[1] Donovan, Brian; İş, dan (2016): New York City TAXI seyahat verileri (2010-2013). University of Illinois at Urbana-Şaign. https://doi.org/10.13012/J8PN93H8

Veri Oluşturucu, kayıtları okuyan ve Azure Event Hubs gönderen bir .NET Core uygulamasıdır. Oluşturucu verileri JSON biçiminde ve tarifeli havayolu verilerinde CSV biçiminde gönderir.

Event Hubs, verileri segmentlere ayırmak için bölümleri kullanır. Bölümler, tüketicinin her bölümü paralel olarak okumasına izin verir. Event Hubs veri gönderdiğinizde, Bölüm anahtarını açıkça belirtebilirsiniz. Aksi takdirde, kayıtlar, hepsini bir kez deneme biçiminde bölümlere atanır.

Bu senaryoda, arttırıldığında verileri ve tarifeli havayolu verileri, belirli bir TAXI cab için aynı bölüm kimliğiyle bitmelidir. Bu, Stream Analytics iki akışı karşılıklı hale geldiğinde paralellik bir derece uygulanmasını sağlar. Arttırıldığında verilerinin bölüm n ' deki bir kayıt, tarifeli havayolu verilerinin bölüm n ' deki bir kayıtla eşleşir.

Veri oluşturucusunda, her iki kayıt türü için ortak veri modeli PartitionKey ,, ve birleşimi olan bir özelliğine sahiptir MedallionHackLicenseVendorId .

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}";
    }

Bu özellik, Event Hubs gönderilirken açık bir bölüm anahtarı sağlamak için kullanılır:

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

Akış işleme

akış işleme işi, birkaç farklı adımla bir SQL sorgusu kullanılarak tanımlanır. İlk iki adım yalnızca iki giriş akışlarından kayıtları seçer.

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
),

Sonraki adım, her akıştaki eşleşen kayıtları seçmek için iki giriş akışını birleştirir.

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
)

Bu sorgu, eşleşen kayıtları benzersiz bir şekilde tanımlayan bir alan kümesindeki kayıtları birleştirir ( PartitionId ve PickupTime ).

Stream Analytics, birleşimler zamana göre yapılır,yani kayıtlar belirli bir zaman penceresi içinde birleştirilir. Aksi takdirde, işin bir eşleşme için süresiz olarak beklemesi gerekebilir. DATEDIFF işlevi, eşleşme için zaman içinde iki eşleşen kaydın ne kadar ayrılabileceğini belirtir.

İşin son adımı, 5 dakikalık bir atlamalı pencereye göre gruplanan, mil başına ortalama ipucunu hesaplar.

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 birkaç Pencereleme işlevisağlar. Bir atlamalı pencere, bu durumda atlama başına 1 dakikalık bir süre içinde sabit bir döneme göre ileriye doğru gider. Sonuç olarak son 5 dakika içinde bir hareketli ortalama hesaplanacak.

burada gösterilen mimaride, yalnızca Stream Analytics işinin sonuçları Cosmos DB kaydedilir. Büyük bir veri senaryosunda, Ham olay verilerini Azure Blob depolama alanına kaydetmek için Event Hubs yakalamayı kullanmayı da düşünün. Ham verilerin tutulması, veri kaynağından yeni Öngörüler türetmek için, daha sonra geçmiş verileriniz üzerinde toplu sorguları çalıştırmanıza olanak sağlar.

Ölçeklenebilirlik konusunda dikkat edilmesi gerekenler

Event Hubs

Event Hubs üretilen iş kapasitesi, üretilen iş birimlericinsinden ölçülür. Otomatik Şişiretkinleştirerek bir olay hub 'ını otomatik ölçeklendirme yaparak, trafiği temel alarak üretilen iş birimlerini otomatik olarak ölçeklendirir.

Stream Analytics

Stream Analytics için, bir işe ayrılan bilgi işlem kaynakları akış birimlerinde ölçülür. İş Paralelleştirilmesi halinde işler ölçeği en iyi Stream Analytics. Bu şekilde, Stream Analytics işi birden çok işlem düğümüne dağıtabilirler.

Event Hubs girişi için, PARTITION BY Stream Analytics işini bölümlemek için anahtar sözcüğünü kullanın. Veriler Event Hubs bölümlerine göre alt kümelere bölünecektir.

Pencereleme işlevleri ve zamana bağlı birleşimler ek SU gerektirir. Mümkün olduğunda, PARTITION BY her bölümün ayrı olarak işlenmesi için kullanın. Daha fazla bilgi için bkz. akış birimlerini anlama ve ayarlama.

Stream Analytics işin tamamını paralel hale getirmek mümkün değilse, bir veya daha fazla paralel adımla başlayarak işi birden çok adıma kesmeyi deneyin. Bu şekilde, ilk adımlar paralel olarak çalıştırılabilir. Örneğin, bu başvuru mimarisinde:

• 1. ve 2. adımlar, SELECT tek bir bölüm içindeki kayıtları seçip basit deyimlerdir.
• 3. Adım iki giriş akışında bölümlenmiş bir JOIN gerçekleştirir. Bu adım, eşleşen kayıtlarla aynı bölüm anahtarını paylaştığından ve her giriş akışında aynı bölüm KIMLIĞINE sahip olduğu garantiden yararlanır.
• Tüm bölümler genelinde 4. adım toplamaları. Bu adım paralelleştirilmedi.

İşteki her bir adıma kaç bölüm atandığını görmek için Stream Analytics iş diyagramını kullanın. Aşağıdaki diyagramda, bu başvuru mimarisinin iş diyagramını gösterir:

Cosmos DB

Cosmos DB için aktarım hızı kapasitesi İstek Birimleri (RU) olarak ölçülür. Cosmos DB kapsayıcısı ölçeğini 10.000 RU'dan önce ölçeklendirmek için, kapsayıcıyı ekleyebilirsiniz ve bölüm anahtarını her belgeye dahil edin.

Bu başvuru mimarisinde, dakikada yalnızca bir kez (atlamalı pencere aralığı) yeni belgeler oluşturulur, bu nedenle aktarım hızı gereksinimleri oldukça düşüktür. Bu nedenle bu senaryoda bölüm anahtarı atamaya gerek yoktur.

İzlemeyle ilgili dikkat edilmesi gerekenler

Herhangi bir akış işleme çözümü ile sistemin performansını ve sistem durumunu izlemek önemlidir. Azure İzleyici mimaride kullanılan Azure hizmetleri için ölçüm ve tanılama günlüklerini toplar. Azure İzleyici Azure platformunda yerleşiktir ve uygulamanıza ek kod gerektirmez.

Aşağıdaki uyarı sinyallerinin herhangi biri, ilgili Azure kaynağının ölçeğini ölçeklendirmeniz gerektiğini gösterir:

• Event Hubs istekleri kısıtlar veya günlük ileti kotasına yakın.
• İş Stream Analytics tutarlı olarak ayrılan Akış Birimlerinin (SU) %80'inden fazlasını kullanır.
• Cosmos DB istekleri kısıtlamaya başlar.

Başvuru mimarisi, panoya dağıtılan özel bir pano Azure portal. Mimariyi dağıtın, panoyu açarak ve Azure portal listesinden seçerek panoyu görüntüleyebilirsiniz. Panolarda özel pano oluşturma ve dağıtma hakkında daha fazla bilgi Azure portal bkz. Program aracılığıyla Azure Panoları oluşturma.

Aşağıdaki görüntüde, Stream Analytics bir saat boyunca Stream Analytics panoyu görebilirsiniz.

Sol altta yer alan panel, ilk 15 dakika boyunca Stream Analytics için SU tüketiminin düşük olduğunu gösterir. İş kararlı bir durumuna ulaştığında bu tipik bir desendir.

Bu Event Hubs, sağ üst panelde gösterilen azaltma isteklerine dikkat ediyor. İstemci SDK'sı bir azaltma hatası aldığında Event Hubs otomatik olarak yeniden denemesi nedeniyle zaman zaman kısıtlandı isteği sorun değildir. Ancak tutarlı azaltma hataları görüyorsanız olay hub'ı daha fazla işleme birimine ihtiyaç gösterir. Aşağıdaki grafikte, işleme birimlerinin ölçeğini gerektiğinde otomatik Event Hubs otomatik olarak şişirme özelliğini kullanan bir test çalıştırması gösterilir.

Otomatik şişirme yaklaşık 06:35 işaretinde etkinleştirildi. Otomatik olarak 3 işleme birimine kadar ölçeklendirilen Event Hubs isteklerin p düşüşlerini görüntülebilirsiniz.

İlginçtir ki bu, iş için su kullanımını artırmanın yan Stream Analytics oldu. Azaltmayla, Event Hubs iş için alımı yapay olarak Stream Analytics azalttı. Aslında bir performans sorununu çözmek başka bir performans sorununu ortaya çıkartır. Bu durumda, iş için ek SU Stream Analytics sorunu çözdü.

Maliyetle ilgili konular

Maliyetleri tahmin etmek için Azure fiyatlandırma hesaplayıcısını kullanın. Bu başvuru mimarisinde kullanılan hizmetler için dikkat edilmesi gereken bazı noktalar burada vetir.

Azure Stream Analytics

Azure Stream Analytics, verileri hizmete işlemesi için gereken akış birimi sayısına (saat 0,11 ABD doları) göre fiyatlandı.

Stream Analytics gerçek zamanlı veya küçük miktarlardaki verilerde işlem değil de pahalı olabilir. Bu kullanım örnekleri için, verileri Azure İşlevleri Logic Apps depolamadan veri deposuna taşımak için Azure Event Hubs veya daha fazla veri kullanmayı göz önünde bulundurabilirsiniz.

Azure Event Hubs ve Azure Cosmos DB

Azure Event Hubs ve Cosmos DB ile ilgili maliyetle ilgili dikkat edilmesi gerekenler için bkz. Azure Databricks mimarisiyle akış işleme.

Çözümü dağıtma

Başvuru uygulamasını dağıtmak ve çalıştırmak için beni okuyun makalesinde GitHub izleyin.

DevOps için dikkat edilmesi gerekenler

• Üretim, geliştirme ve test ortamları için ayrı kaynak grupları oluşturun. Ayrı kaynak grupları dağıtımları yönetmeyi, test dağıtımlarını silmeyi ve erişim haklarını atamayı kolaylaştırır.

• Kod Azure Resource Manager (IaC) işleminin ardından Azure kaynaklarını dağıtmak için azure şablonunu kullanın. Şablonlarla, dağıtımları Azure DevOps Services veyadiğer CI/CD çözümlerini kullanarak otomatik hale getirirsiniz.

• Her iş yükünü ayrı bir dağıtım şablonuna koyarak kaynakları kaynak denetim sistemlerinde depolar. Şablonları CI/CD işleminin bir parçası olarak birlikte veya tek tek dağıtarak otomasyon sürecini kolaylaştırabilirsiniz.

  Bu mimaride Azure Event Hubs, Log Analytics ve Cosmos DB tek bir iş yükü olarak tanımlanır. Bu kaynaklar tek bir ARM şablonuna dahildir.

• İş yüklerinizi hazırlamayı göz önünde bulundurarak. Bir sonraki aşamaya taşınmadan önce çeşitli aşamalara dağıtın ve doğrulama denetimlerini her aşamada çalıştırın. Bu şekilde, üretim ortamlarınızı yüksek oranda denetimli bir şekilde güncelleştirmeler anında iletir ve beklentisiz dağıtım sorunlarını en aza saldırabilirsiniz.

• Akış işleme Azure İzleyici performansını analiz etmek için Azure İzleyici kullanmayı göz önünde bulundurabilirsiniz. Daha fazla bilgi için bkz. İzleme Azure Databricks.

Daha fazla bilgi için DevOps Framework'te Microsoft Azure Well-Architected bölümüne bakın.

İlgili kaynaklar

Aynı teknolojilerden bazıları kullanılarak belirli çözümleri gösteren aşağıdaki Azure örnek senaryolarını gözden geçirmek iyi olabilir:

• İnşaat sektöründe IoT ve veri analizi
• Gerçek zamanlı sahtekarlık algılama