Azure Stream Analytics ile akış işleme

Bu başvuru mimarisi uçtan uca akış işleme işlem hattını gösterir. İşlem hattı iki kaynaktan veri alır, iki akıştaki kayıtları ilişkilendirip bir zaman penceresinde sıralı ortalama hesaplar. Sonuçlar daha fazla analiz için depolanır.

Bu mimari için bir başvuru uygulaması GitHub'da kullanılabilir.

Mimari

Bu mimarinin bir Visio dosyasını indirin.

İş Akışı

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

Veri kaynakları. Bu mimaride, gerçek zamanlı olarak veri akışları oluşturan iki veri kaynağı vardır. İlk akış sürüş bilgilerini, ikinci akış ise ücret bilgilerini içerir. Başvuru mimarisi, bir dizi statik dosyadan okuyan ve verileri Event Hubs'a gönderen bir sanal veri oluşturucu içerir. Gerçek bir uygulamada, veri kaynakları taksilere yüklenen cihazlar olacaktır.

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

Azure Stream Analytics. Stream Analytics bir olay işleme altyapısıdır. Stream Analytics işi iki olay hub'ından veri akışlarını okur ve akış işleme gerçekleştirir.

Azure Cosmos DB. Stream Analytics işinin çıktısı, Azure Cosmos DB belge veritabanına JSON belgeleri olarak yazılan bir dizi kayıttır.

Microsoft Power BI. Power BI, iş içgörüleri için verileri analiz etmeye yönelik bir iş analizi araçları paketidir. Bu mimaride Azure Cosmos DB'deki verileri yükler. Bu, kullanıcıların toplanan tüm geçmiş verileri analiz etmesine olanak tanır. Ayrıca verilerin gerçek zamanlı bir görünümü için sonuçları doğrudan Stream Analytics'ten Power BI'a aktarabilirsiniz. Daha fazla bilgi için bkz. Power BI’da gerçek zamanlı akış yapma.

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

Senaryo ayrıntıları

Senaryo: Bir taksi şirketi her taksi yolculuğu hakkında veri toplar. Bu senaryo için veri gönderen iki ayrı cihaz olduğunu varsayıyoruz. Taksinin her yolculuk hakkında bilgi gönderen bir ölçümü vardır: süre, mesafe, teslim alma ve bırakma konumları. Ayrı bir cihaz müşterilerden gelen ödemeleri kabul eder ve ücretler hakkında veri gönderir. Taksi şirketi, eğilimleri tespit etmek için kilometre başına ortalama ipucunu gerçek zamanlı olarak hesaplamak istiyor.

Olası kullanım örnekleri

Bu çözüm, perakende senaryosu için iyileştirilmiştir.

Veri alımı

Bir veri kaynağının benzetimini yapmak için bu başvuru mimarisi New York City Taxi Data veri kümesini ^kullanır[1]. Bu veri kümesi, dört yıllık bir dönemde (2010-2013) New York'taki taksi yolculukları hakkındaki verileri içerir. İki tür kayıt içerir: sürüş verileri ve ücret verileri. Yolculuk verileri seyahat süresini, seyahat mesafesini, teslim alma ve bırakma konumunu içerir. Ücret verileri, ücret, vergi ve ipucu tutarlarını içerir. Her iki kayıt türündeki yaygın alanlar arasında madalyon numarası, hack lisansı ve satıcı kimliği yer alır. Bu üç alan birlikte benzersiz olarak bir taksi ve bir sürücü tanımlar. Veriler CSV biçiminde depolanır.

[1] Donovan, Brian; Work, Dan (2016): New York City Taxi Trip Data (2010-2013). Urbana-Champaign'deki Illinois Üniversitesi. https://doi.org/10.13012/J8PN93H8

Veri oluşturucu, kayıtları okuyan ve Azure Event Hubs'a gönderen bir .NET Core uygulamasıdır. Oluşturucu, sürüş verilerini JSON biçiminde ve ücret verilerini 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 okumasına olanak sağlar. Event Hubs'a veri gönderdiğinizde bölüm anahtarını açıkça belirtebilirsiniz. Aksi takdirde, kayıtlar bölümlere hepsini bir kez deneme biçiminde atanır.

Bu özel senaryoda, sürüş verileri ve ücret verileri belirli bir taksi için aynı bölüm kimliğine sahip olmalıdır. Bu, Stream Analytics'in iki akışla bağıntılı olduğunda bir derece paralellik uygulamasına olanak tanır. Yolculuk verilerinin n. bölümündeki bir kayıt, ücret verilerinin n. bölümündeki bir kayıtla eşleşecektir.

Veri oluşturucuda, her iki kayıt türü için ortak veri modelinin , ve birleştirmesi Medallionolan bir PartitionKey özelliği VendorIdvardır. HackLicense

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'a gönderirken 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ıma sahip bir SQL sorgusu kullanılarak tanımlanır. İlk iki adım, iki giriş akışından kayıtları seçmeniz yeterlidir.

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ıştan 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ı (PartitionId ve PickupTime) benzersiz olarak tanımlayan bir alan kümesindeki kayıtları birleştirir.

Stream Analytics'te birleştirmeler zamansaldır, yani kayıtlar belirli bir zaman aralığı içinde birleştirilir. Aksi takdirde, işin bir eşleşme için süresiz olarak beklemesi gerekebilir. DATEDIFF işlevi, eşleşen iki kaydın bir eşleşme için ne kadar süreyle ayrılabileceğini belirtir.

İşin son adımı kilometre başına ortalama ipucunu hesaplar ve 5 dakikalık bir atlama penceresine göre gruplandırılır.

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 çeşitli pencereleme işlevleri sağlar. Atlamalı pencere, sabit bir süre ileri doğru ilerler ve bu durumda atlama başına 1 dakika. Sonuç, son 5 dakika içindeki hareketli ortalamayı hesaplamaktır.

Burada gösterilen mimaride yalnızca Stream Analytics işinin sonuçları Azure Cosmos DB'ye kaydedilir. Büyük bir veri senaryosu için event hubs capture kullanarak ham olay verilerini Azure Blob depolamaya kaydetmeyi de göz önünde bulundurun. Ham verileri tutmak, verilerden yeni içgörüler elde etmek için geçmiş verileriniz üzerinde daha sonra toplu sorgular çalıştırmanıza olanak sağlar.

Dikkat edilmesi gereken noktalar

Bu önemli noktalar, bir iş yükünün kalitesini artırmak için kullanılabilecek bir dizi yol gösteren ilke olan Azure İyi Tasarlanmış Çerçeve'nin yapı taşlarını uygular. Daha fazla bilgi için bkz . Microsoft Azure İyi Tasarlanmış Çerçeve.

Ölçeklenebilirlik

Event Hubs

Event Hubs'ın aktarım hızı kapasitesi aktarım hızı birimleriyle ölçülür. Otomatik şişirme özelliğini etkinleştirerek olay hub'ını otomatik olarak ölçeklendirerek aktarım hızı birimlerini trafiğe göre yapılandırılan en yüksek değere kadar ölçeklendirin.

Stream Analytics

Stream Analytics için bir işe ayrılan bilgi işlem kaynakları Akış Birimleri cinsinden ölçülür. Stream Analytics işleri, iş paralelleştirilebiliyorsa en iyi şekilde ölçeklendirilir. Bu şekilde Stream Analytics işi birden çok işlem düğümü arasında dağıtabilir.

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

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

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

• Adım 1 ve 2, tek bir bölümdeki kayıtları seçen basit SELECT deyimlerdir.
• 3. Adım, iki giriş akışında bölümlenmiş birleştirme gerçekleştirir. Bu adım, eşleşen kayıtların aynı bölüm anahtarını paylaşması gerçeğinden yararlanır ve bu nedenle her giriş akışında aynı bölüm kimliğine sahip olması garanti edilir.
• 4. adım, tüm bölümlerde toplanır. Bu adım paralelleştirilemez.

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

Azure Cosmos DB

Azure Cosmos DB için aktarım hızı kapasitesi İstek Birimleri (RU) cinsinden ölçülür. Azure Cosmos DB kapsayıcısını 10.000 RU'yu aşacak şekilde ölçeklendirmek için kapsayıcıyı oluştururken bir bölüm anahtarı belirtmeniz ve bölüm anahtarını her belgeye eklemeniz gerekir.

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

İzleme

Herhangi bir akış işleme çözümünde 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ızda ek kod gerektirmez.

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

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

Başvuru mimarisi, Azure portalına dağıtılan özel bir pano içerir. Mimariyi dağıttığınızda, Azure portalını açıp pano listesinden öğesini seçerek TaxiRidesDashboard panoyu görüntüleyebilirsiniz. Azure portalında özel pano oluşturma ve dağıtma hakkında daha fazla bilgi için bkz . Program aracılığıyla Azure Panoları oluşturma.

Aşağıdaki görüntüde, Stream Analytics işi yaklaşık bir saat boyunca çalıştırıldıktan sonra pano gösterilmektedir.

Sol alt taraftaki panel, Stream Analytics işi için SU tüketiminin ilk 15 dakika boyunca tırmandığını ve ardından seviye atlandığını gösterir. İş sabit bir duruma ulaştığından bu tipik bir desendir.

Event Hubs'ın sağ üst panelde gösterilen istekleri azalttığını göreceksiniz. Event Hubs istemci SDK'sı azaltma hatası aldığında otomatik olarak yeniden denendiğinden, zaman zaman kısıtlanan istek sorun oluşturmaz. Ancak tutarlı azaltma hataları görürseniz olay hub'ına daha fazla aktarım hızı birimi gerekir. Aşağıdaki grafikte Event Hubs otomatik olarak şişirme özelliğinin kullanıldığı ve gerektiğinde aktarım hızı birimlerinin ölçeğini otomatik olarak genişleten bir test çalıştırması gösterilmektedir.

Otomatik şişirme yaklaşık 06:35 işaretinde etkinleştirildi. Event Hubs otomatik olarak 3 aktarım hızı birimine kadar ölçeklendirildiğinden kısıtlanmış isteklerde p düşüşünü görebilirsiniz.

İlginçtir ki bu, Stream Analytics işinde SU kullanımını artırmanın yan etkisi oldu. Event Hubs, azaltma işlemiyle Stream Analytics işinin alım oranını yapay olarak azaltıyordu. Aslında bir performans sorununu çözmenin başka bir performans sorununa neden olması yaygın bir durumdur. Bu durumda, Stream Analytics işi için ek SU tahsis etmek sorunu çözdü.

Maliyet iyileştirme

Maliyet iyileştirmesi, gereksiz giderleri azaltmanın ve operasyonel verimlilikleri iyileştirmenin yollarını aramaktır. Daha fazla bilgi için bkz . Maliyet iyileştirme sütununa genel bakış.

Maliyetleri tahmin etmek için Azure fiyatlandırma hesaplayıcısını kullanın. Bu başvuru mimarisinde kullanılan hizmetlerle ilgili dikkat edilmesi gereken bazı noktalar aşağıdadır.

Azure Stream Analytics

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

Stream Analytics, verileri gerçek zamanlı veya az miktarda veriyle işlemezseniz pahalı olabilir. Bu kullanım örnekleri için verileri Azure Event Hubs'dan bir veri deposuna taşımak için Azure İşlevleri veya Logic Apps kullanmayı göz önünde bulundurun.

Azure Event Hubs ve Azure Cosmos DB

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

DevOps

• Ü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 Olarak Altyapı (IaC) İşlemi'ni izleyen Azure kaynaklarını dağıtmak için Azure Resource Manager şablonunu kullanın. Şablonlarla, Azure DevOps Services veya diğer CI/CD çözümlerini kullanarak dağıtımları otomatikleştirmek daha kolaydır.

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

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

• İş yüklerinizi hazırlamayı göz önünde bulundurun. Çeşitli aşamalara dağıtın ve sonraki aşamaya geçmeden önce her aşamada doğrulama denetimleri çalıştırın. Bu şekilde, güncelleştirmeleri üretim ortamlarınıza yüksek denetimli bir şekilde gönderebilir ve tahmin edilmeyen dağıtım sorunlarını en aza indirebilirsiniz.

• Akış işleme işlem hattınızın performansını analiz etmek için Azure İzleyici'yi kullanmayı göz önünde bulundurun. Daha fazla bilgi için bkz . Azure Databricks'i izleme.

Daha fazla bilgi için bkz. Microsoft Azure İyi Tasarlanmış Çerçeve'deki operasyonel mükemmellik sütunu.

Bu senaryoyu dağıtın

Başvuru uygulamasını dağıtmak ve çalıştırmak için GitHub benioku dosyasındaki adımları izleyin.

İlgili kaynaklar

Aynı teknolojilerden bazılarını kullanarak belirli bir çözümü gösteren aşağıdaki Azure örnek senaryoyu gözden geçirmek isteyebilirsiniz:

• Gerçek zamanlı sahtekarlık algılama