Aracılığıyla paylaş

Olay çoğaltma görevleri desenleri

  • Makale

Federasyona genel bakış ve çoğaltıcı işlevlerine genel bakış , çoğaltma görevlerinin rasyonalasyonunu ve temel öğelerini açıklar ve bu makaleye devam etmeden önce bunları öğrenmeniz önerilir.

Bu makalede, genel bakış bölümünde vurgulanan çeşitli desenler için uygulama kılavuzunu ayrıntılı olarak açıklıyoruz.

Çoğaltma

Çoğaltma düzeni olayları bir Olay Hub'ından diğerine veya bir Olay Hub'ından Service Bus kuyruğu gibi başka bir hedefe kopyalar. Olaylar, olay yükünde herhangi bir değişiklik yapılmadan iletilir.

Bu desenin uygulanması, Event Hubs ile Service Bus örnekleri arasında Event Hubs ile Olay çoğaltması arasında olay çoğaltması ve Apache Kafka aracısından Event Hubs'a veri çoğaltmaya özgü durum için Event Hubs ile Apache Kafka MirrorMaker kullanma öğreticisi kapsamındadır.

Akışlar ve düzen koruması

Azure İşlevleri veya Azure Stream Analytics aracılığıyla çoğaltma, bir kaynak Olay Hub'ının tam 1:1 kopyasının hedef Olay Hub'ına oluşturulmasını sağlamayı hedeflemez, ancak uygulamanın gerektirdiği olayların göreli sırasını korumaya odaklanır. Uygulama, ilgili olayları aynı bölüm anahtarıyla gruplandırarak bunu bildirir ve Event Hubs aynı bölüm anahtarına sahip iletileri aynı bölümde sıralı olarak düzenler.

Önemli

"Uzaklık" bilgileri her Olay Hub'ı için benzersizdir ve aynı olayların uzaklıkları Olay Hub'ı örnekleri arasında farklılık gösterir. Kopyalanan olay akışındaki bir konumu bulmak için zamana dayalı uzaklıkları kullanın ve hizmet tarafından atanan yayılan meta verilerine bakın.

Zaman tabanlı uzaklıklar, alıcınızı belirli bir zamanda başlatır:

  • EventPosition.FromStart() - Tutulan tüm verileri yeniden okuyun.
  • EventPosition.FromEnd() - Bağlantı zamanından itibaren tüm yeni verileri okuyun.
  • EventPosition.FromEnqueuedTime(dateTime) - Belirli bir tarih ve saatten başlayan tüm veriler.

EventProcessor'da, EventProcessorOptions üzerindeki InitialOffsetProvider aracılığıyla konumu ayarlarsınız. Diğer alıcı API'leriyle, konum oluşturucudan geçirilir.

Azure İşlevleri tabanlı kılavuzda kullanılan örnekler olarak sağlanan önceden oluşturulmuş çoğaltma işlevi yardımcıları, kaynak bölümden alınan aynı bölüm anahtarına sahip olay akışlarının hedef Olay Hub'ına özgün akışta bir toplu iş olarak ve aynı bölüm anahtarıyla gönderilmesini sağlar.

Kaynak ve hedef Olay Hub'ının bölüm sayısı aynıysa, hedefteki tüm akışlar kaynakta olduğu gibi aynı bölümlere eşlenir. Aşağıda açıklanan diğer desenlerden bazılarında önemli olan bölüm sayısı farklıysa eşleme farklılık gösterir, ancak akışlar her zaman birlikte ve sırayla tutulur.

Farklı akışlara veya hedef bölümde bölüm anahtarı olmayan bağımsız olaylara ait olayların göreli sırası her zaman kaynak bölümden farklı olabilir.

Hizmet tarafından atanan meta veriler

Kaynak Olay Hub'ından alınan bir olayın hizmet tarafından atanan meta verileri, özgün sıra süresi, sıra numarası ve uzaklık, hedef Olay Hub'ında hizmet tarafından atanan yeni değerlerle değiştirilir, ancak yardımcı işlevler, örneklerimizde sağlanan çoğaltma görevleriyle özgün değerler kullanıcı özelliklerinde korunur: repl-enqueue-time (ISO8601 dize), repl-sequence, repl-offset.

Bu özellikler dize türündedir ve ilgili özgün özelliklerin dizeleştirilmiş değerini içerir. Olay birden çok kez iletilirse, hemen kaynağın hizmet tarafından atanan meta verileri zaten var olan özelliklere eklenir ve değerler noktalı virgülle ayrılır.

Yük devretme

Çoğaltmayı olağanüstü durum kurtarma amacıyla, Event Hubs hizmetindeki bölgesel kullanılabilirlik olaylarına karşı veya ağ kesintilerine karşı korumak için kullanıyorsanız, bu tür bir hata senaryosunda üreticilere ve/veya tüketicilere ikincil uç noktayı kullanmaları için bir Olay Hub'ından diğerine yük devretme gerçekleştirmeniz gerekir.

Tüm yük devretme senaryolarında, ad alanlarının gerekli öğelerinin yapısal olarak aynı olduğu varsayılır, yani Event Hubs ve Tüketici Grupları aynı şekilde adlandırılır ve paylaşılan erişim imzası kuralları ve/veya rol tabanlı erişim denetimi kuralları aynı şekilde ayarlanır. Ad alanlarını taşıma yönergelerini izleyerek ve temizleme adımını atlayarak ikincil ad alanı oluşturabilir (ve güncelleştirebilirsiniz).

Üreticileri ve tüketicileri geçiş yapmaya zorlamak için, ulaşılıp güncelleştirilebilecek bir konumda arama için hangi ad alanının kullanılacağına ilişkin bilgileri kullanmanız gerekir. Üreticiler veya tüketiciler sık veya kalıcı hatalarla karşılaşırsa bu konuma danışmalı ve yapılandırmalarını ayarlamalıdır. Bu yapılandırmayı paylaşmanın birçok yolu vardır, ancak aşağıdaki ikisine işaret ediyoruz: DNS ve dosya paylaşımları.

DNS tabanlı yük devretme yapılandırması

Aday yaklaşımlardan biri, denetlediğiniz bir DNS'te DNS SRV kayıtlarındaki bilgileri tutmak ve ilgili Olay Hub'ı uç noktalarına işaret etmektir.

Önemli

Event Hubs'ın uç noktalarının doğrudan CNAME kayıtlarıyla diğer adla kullanılmasına izin vermediğini unutmayın. Bu, DNS'yi ip adresi bilgilerini doğrudan çözümlemek yerine uç nokta adresleri için dayanıklı bir arama mekanizması olarak kullanacağınız anlamına gelir.

Etki alanının example.com sahibi olduğunuzu ve uygulamanız için bir bölge test.example.comolduğunu varsayalım. İki alternatif Event Hubs için artık iki iç içe bölge ve her birinde bir SRV kaydı oluşturacaksınız.

SRV kayıtları, iki uç nokta kaydıyla _azure_eventhubs._amqp önekli ve ayrı olan ortak kuralı izler: Biri 5671 numaralı bağlantı noktasında AMQP-over-TLS için, diğeri de 443 numaralı bağlantı noktasında AMQP-over-WebSockets için, her ikisi de bölgeye karşılık gelen ad alanının Event Hubs uç noktasına işaret eder.

Bölge SRV kaydı
eh1.test.example.com _azure_servicebus._amqp.eh1.test.example.com
1 1 5671 eh1-test-example-com.servicebus.windows.net
2 2 443 eh1-test-example-com.servicebus.windows.net
eh2.test.example.com _azure_servicebus._amqp.eh2.test.example.com
1 1 5671 eh2-test-example-com.servicebus.windows.net
2 2 443 eh2-test-example-com.servicebus.windows.net

Uygulamanızın bölgesinde, birincil Olay Hub'ınıza karşılık gelen alt bölgeye işaret eden bir CNAME girdisi oluşturacaksınız:

CNAME kaydı Diğer ad
eventhub.test.example.com eh1.test.example.com

CNAME ve SRV kayıtlarını açıkça sorgulamaya olanak tanıyan bir DNS istemcisi kullanarak (Java ve .NET'in yerleşik istemcileri yalnızca IP adreslerine adların basit bir şekilde çözümlenmesine izin verir), daha sonra istediğiniz uç noktayı çözümleyebilirsiniz. Örneğin DnsClient.NET ile arama işlevi şu şekildedir:

static string GetEventHubName(string aliasName)
{
    const string SrvRecordPrefix = "_azure_eventhub._amqp.";
    LookupClient lookup = new LookupClient();

    return (from CNameRecord alias in (lookup.Query(aliasName, QueryType.CNAME).Answers)
            from SrvRecord srv in lookup.Query(SrvRecordPrefix + alias.CanonicalName, QueryType.SRV).Answers
            where srv.Port == 5671
            select srv.Target).FirstOrDefault()?.Value.TrimEnd('.');
}

İşlev, yukarıda gösterildiği gibi şu anda CNAME ile diğer adı verilen bölgenin 5671 numaralı bağlantı noktası için kaydedilen hedef ana bilgisayar adını döndürür.

Yük devretme gerçekleştirmek için CNAME kaydının düzenlenmesi ve alternatif bölgeye işaretlenmesi gerekir.

DNS ve özellikle Azure DNS kullanmanın avantajı, Azure DNS bilgilerinin genel olarak çoğaltılması ve bu nedenle tek bölgeli kesintilere karşı dayanıklı olmasıdır.

Bu yordam, Event Hubs Geo-DR'nin çalışma şekline benzer, ancak tamamen kendi denetiminizdedir ve etkin/etkin senaryolarla da çalışır.

Dosya paylaşımı tabanlı yük devretme yapılandırması

Uç nokta bilgilerini paylaşmak için DNS kullanmanın en basit alternatifi, birincil uç noktanın adını düz metin dosyasına koymak ve dosyayı kesintilere karşı sağlam ve güncelleştirmelere izin veren bir altyapıdan sunmaktır.

Genel kullanılabilirlik ve içerik çoğaltma ile yüksek oranda kullanılabilir bir web sitesi altyapısını zaten çalıştırıyorsanız, bu tür bir dosyayı oraya ekleyin ve bir anahtar gerekiyorsa dosyayı yeniden yayımlayın.

Dikkat

Gizli diziler dahil olmak üzere tam bir bağlantı dizesi değil, yalnızca uç nokta adını bu şekilde yayımlamanız gerekir.

Tüketicilerin yük devretmesi için dikkat edilmesi gereken ek noktalar

Event Hub tüketicileri için yük devretme stratejisiyle ilgili dikkat edilmesi gereken diğer noktalar olay işlemcisinin gereksinimlerine bağlıdır.

Yedekleme verilerinden veritabanları da dahil olmak üzere bir sistemin yeniden oluşturulmasını gerektiren bir olağanüstü durum varsa ve veritabanları olay hub'ında tutulan olaylardan doğrudan veya ara işleme yoluyla besleniyorsa, yedeklemeyi geri yükler ve ardından, özgün sistemin yok edildiği andan değil, veritabanı yedeklemesinin oluşturulduğu andan itibaren olayları sisteme yeniden yürütmeye başlamak istersiniz.

Bir hata yalnızca sistemin bir dilimini veya ulaşılamaz hale gelen tek bir Olay Hub'ını etkiliyorsa, büyük olasılıkla işlemenin kesildiği konumdan olayları işlemeye devam etmek istersiniz.

Senaryolardan birini gerçekleştirmek ve ilgili Azure SDK'nızın olay işlemcisini kullanmak için yeni bir denetim noktası deposu oluşturacak ve işlemeye devam etmek istediğiniz zaman damgasını temel alan bir ilk bölüm konumu sağlayacaksınız.

Geçiş yaptığınız Olay Hub'ının denetim noktası deposuna hala erişiminiz varsa, yukarıda açıklanan yayılan meta veriler zaten işlenmiş olan olayları atlamanıza ve son kaldığınız yerden tam olarak sürdürmenize yardımcı olur.

Adres Mektup Birleştirme

Birleştirme düzeninde bir hedefe işaret eden bir veya daha fazla çoğaltma görevi vardır; büyük olasılıkla normal üreticiler de aynı hedefe olay gönderir.

Bu patter'lerin varyasyonları şunlardır:

  • İki veya daha fazla çoğaltma işlevi, olayları ayrı kaynaklardan eşzamanlı olarak alır ve aynı hedefe gönderir.
  • Hedef doğrudan üreticiler tarafından da kullanılırken bir kaynaktan olay alan bir çoğaltma işlevi daha.
  • Önceki desen, iki veya daha fazla Olay Hub'ı arasında yansıtılır ve olayların nerede üretildiğine bakılmaksızın bu Event Hubs'ların aynı akışları içermesine neden olur.

İlk iki desen varyasyonu önemsizdir ve düz çoğaltma görevlerinden farklı değildir.

Son senaryo, zaten çoğaltılmış olayların yeniden çoğaltılmasını dışlamanızı gerektirir. Teknik, EventHubToEventHubMerge örneğinde gösterilmiştir ve açıklanmıştır.

Düzenleyici

Düzenleyici düzeni çoğaltma deseni üzerinde oluşturulur, ancak iletiler iletilmeden önce değiştirilir.

Bu tür değişikliklere örnek olarak şunlar verilebilir:

  • Kod dönüştürme - Olay içeriği ("gövde" veya "yük" olarak da adlandırılır) Apache Avro biçimi veya özel bir serileştirme biçimi kullanılarak kodlanmış kaynaktan geliyorsa, ancak hedefe sahip sistemin beklentisi içeriğin JSON kodlanmış olmasıysa, kod dönüştürme çoğaltma görevi önce Apache Avro'dan bellek içi nesne grafiğine yükü seri durumdan çıkararak bu grafiği JSON'a seri hale getirecektir biçimine bakın. Kod dönüştürme, içerik sıkıştırma ve sıkıştırmayı açma görevlerini de içerir.
  • Dönüştürme - Yapılandırılmış veriler içeren olaylar, aşağı akış tüketicilerinin daha kolay kullanımı için bu verilerin yeniden şekillenmesi gerektirebilir. Bu, iç içe yapıları düzleştirme, gereksiz veri öğelerini ayıklama veya yükü belirli bir şemaya tam olarak uyacak şekilde yeniden şekillendirme gibi işleri içerebilir.
  • Toplu işleme - Olaylar bir kaynaktan toplu olarak (tek bir aktarımda birden çok olay) alınabilir, ancak tek başına bir hedefe iletilmesi gerekir (veya tam tersi). Bu nedenle bir görev, tek bir giriş olayı aktarımına dayalı olarak birden çok olayı iletebilir veya daha sonra birlikte aktarılan bir dizi olayı toplayabilir.
  • Doğrulama - Dış kaynaklardan gelen olay verilerinin iletilmeden önce genellikle bir dizi kuralla uyumlu olup olmadığının denetlenip denetlenmediğine bakılmalıdır. Kurallar şemalar veya kod kullanılarak ifade edilebilir. Uyumlu olmadığı belirlenen olaylar bırakılabilir ve sorun günlüklerde belirtilebilir veya bunları daha fazla işlemek için özel bir hedef hedefe iletilebilir.
  • Zenginleştirme - Bazı kaynaklardan gelen olay verileri, hedef sistemlerde kullanılabilir olması için daha fazla bağlamla zenginleştirme gerektirebilir. Bu, başvuru verilerini aramayı ve bu verileri olayla birlikte eklemeyi veya çoğaltma görevi tarafından bilinen ancak olaylara dahil olmayan kaynak hakkında bilgi eklemeyi içerebilir.
  • Filtreleme - Bir kaynaktan gelen bazı olayların bazı kurallar temelinde hedeften engellenmesi gerekebilir. Filtre, olayı bir kurala göre sınar ve olay kuralla eşleşmezse olayı bırakır. Belirli ölçütleri gözlemleyerek ve aynı değerlere sahip sonraki olayları bırakarak yinelenen olayları filtrelemek bir filtreleme biçimidir.
  • Şifreleme - Çoğaltma görevinin kaynaktan gelen içeriğin şifresini çözmesi ve/veya hedefe doğru iletilen içeriği şifrelemesi ve/veya olayda taşınan imzaya göre içerik ve meta verilerin bütünlüğünü doğrulaması veya böyle bir imza eklemesi gerekebilir.
  • Kanıtlama - Çoğaltma görevi, olayın belirli bir kanaldan veya belirli bir zamanda alındığını kanıtlayan bir olaya dijital imzayla korunan meta veriler ekleyebilir.
  • Zincirleme - Çoğaltma görevi, akışın bütünlüğünün korunması ve eksik olayların algılanması gibi olay akışlarına imza uygulayabilir.

Dönüştürme, toplu işlem ve zenginleştirme desenleri genellikle en iyi şekilde Azure Stream Analytics işleriyle uygulanır.

Tüm bu desenler, olayları almak için Event Hubs Tetikleyicisi ve bunları teslim etmek için Event Hub çıkış bağlaması kullanılarak Azure İşlevleri kullanılarak uygulanabilir.

Yönlendirme

Yönlendirme düzeni çoğaltma deseni üzerinde derlenmiş, ancak bir kaynak ve bir hedef yerine, çoğaltma görevinin birden çok hedefi vardır ve burada C# dilinde gösterilmiştir:

[FunctionName("EH2EH")]
public static async Task Run(
    [EventHubTrigger("source", Connection = "EventHubConnectionAppSetting")] EventData[] events,
    [EventHub("dest1", Connection = "EventHubConnectionAppSetting")] EventHubClient output1,
    [EventHub("dest2", Connection = "EventHubConnectionAppSetting")] EventHubClient output2,
    ILogger log)
{
    foreach (EventData eventData in events)
    {
        // send to output1 and/or output2 based on criteria
        EventHubReplicationTasks.ConditionalForwardToEventHub(input, output1, log, (eventData) => {
            return ( inputEvent.SystemProperties.SequenceNumber%2==0 ) ? inputEvent : null;
        });
        EventHubReplicationTasks.ConditionalForwardToEventHub(input, output2, log, (eventData) => {
            return ( inputEvent.SystemProperties.SequenceNumber%2!=0 ) ? inputEvent : null;
        });
    }
}

Yönlendirme işlevi, ileti meta verilerini ve/veya ileti yükünü dikkate alır ve ardından gönderilecek kullanılabilir hedeflerden birini seçer.

Azure Stream Analytics'te, aynı işlemi birden çok çıkış tanımlayıp çıkış başına sorgu yürüterek elde edebilirsiniz.

select * into dest1Output from inputSource where Info = 1
select * into dest2Output from inputSource where Info = 2

Günlük projeksiyonu

Günlük projeksiyon deseni, olay akışını dizine alınan bir veritabanına düzleştirilir ve olaylar veritabanında kayıt haline gelir. Olaylar genellikle aynı koleksiyona veya tabloya eklenir ve Event Hub bölüm anahtarı, kaydı benzersiz hale getirmek için birincil anahtarın bir parçası olur.

Günlük projeksiyonu, olay verilerinizin zaman serisi tarihçisini veya her bölüm anahtarı için yalnızca en son olayın korunduğu sıkıştırılmış bir görünüm oluşturabilir. Hedef veritabanının şekli nihai olarak size ve uygulamanızın ihtiyaçlarına göre ayarlanır. Bu desene "olay kaynağını belirleme" de denir.

İpucu

Azure Stream Analytics'te Azure SQL Veritabanı ve Azure Cosmos DB'de kolayca günlük projeksiyonları oluşturabilirsiniz ve bu seçeneği tercih etmelisiniz.

Aşağıdaki Azure İşlevi, bir Azure Cosmos DB koleksiyonuna sıkıştırılmış bir Olay Hub'ının içeriğini projelendirmektedir.

[FunctionName("Eh1ToCosmosDb1Json")]
[ExponentialBackoffRetry(-1, "00:00:05", "00:05:00")]
public static async Task Eh1ToCosmosDb1Json(
    [EventHubTrigger("eh1", ConsumerGroup = "Eh1ToCosmosDb1", Connection = "Eh1ToCosmosDb1-source-connection")] EventData[] input,
    [CosmosDB(databaseName: "SampleDb", collectionName: "foo", ConnectionStringSetting = "CosmosDBConnection")] IAsyncCollector<object> output,
    ILogger log)
{
    foreach (var ev in input)
    {
        if (!string.IsNullOrEmpty(ev.SystemProperties.PartitionKey))
        {
            var record = new
            {
                id = ev.SystemProperties.PartitionKey,
                data = JsonDocument.Parse(ev.Body),
                properties = ev.Properties
            };
            await output.AddAsync(record);
        }
    }
}

Sonraki adımlar