イベントレプリケーションタスクのパターン

[アーティクル]
06/01/2023

フェデレーションの概要とレプリケーター関数の概要に関するページでは、レプリケーションタスクの原理と基本要素について説明しています。この記事を読み進める前にこれらについてよく理解しておくことをお勧めします。

この記事では、概要セクションで強調されているいくつかのパターンの実装ガイダンスについて詳しく説明します。

レプリケーション

レプリケーションパターンでは、あるイベントハブから次のものへ、またはイベントハブから、Service Bus キューなどの他の宛先にイベントをコピーします。イベントは、イベントペイロードを変更することなく転送されます。

このパターンの実装は、「Event Hubs 間のイベントレプリケーション」および「Event Hubs と Service Bus 間のイベントレプリケーション」のサンプル、および Apache Kafka ブローカーから Event Hubs にデータをレプリケートする特定のケースについては、「Event Hubs で Apache Kafka MirrorMaker を使用する」のチュートリアルに記載されています。

ストリームと順序の維持

Azure Functions または Azure Stream Analytics を使用するレプリケーションでは、ソースイベントハブからターゲットイベントハブへの正確な 1 対 1 のクローン作成を確保することを目的としているのではなく、アプリケーションで必要な場所にイベントの相対順序を維持することに重点を置いています。アプリケーションにより、同じパーティションキーを持つ関連するイベントをグループ化することでこのことが伝達され、Event Hubs により、同じパーティションキーを持つメッセージが同じパーティション内に順番に配置されます。

重要

"オフセット" 情報はイベントハブごとに一意であり、同じイベントのオフセットはイベントハブインスタンス間で異なります。コピーされたイベントストリームで位置を見つけるには、時間ベースのオフセットを使用して、伝達されたサービスによって割り当てられたメタデータを参照します。

時間ベースのオフセットでは、特定の時点でレシーバーを開始します。

EventPosition.FromStart() - 保持されているすべてのデータを再度読み取ります。
EventPosition.FromEnd() - 接続の時点からすべての新しいデータを読み取ります。
EventPosition.FromEnqueuedTime(dateTime) - 指定した日付と時刻から始まるすべてのデータ。

EventProcessor では、EventProcessorOptions で InitialOffsetProvider を使用して位置を設定します。他のレシーバー API では、位置がコンストラクター経由で渡されます。

事前に構築されたレプリケーション関数のヘルパー (Azure Functions ベースのガイダンスで使用されるサンプルとして提供) により、ソースパーティションから取得された同じパーティションキーを持つイベントストリームが、同じパーティションキーで元のストリームのバッチとしてターゲットイベントハブに送信されることが保証されます。

ソースおよびターゲットイベントハブのパーティション数が同一であれば、ターゲット内のすべてのストリームが、ソースの場合と同じパーティションにマップされます。パーティション数が異なる場合 (以下に示す別のパターンのいくつかでは重要)、マッピングは異なりますが、ストリームは常に順序どおりにまとめられます。

異なるストリームに属するイベント、またはターゲットパーティションのパーティションキーがない独立したイベントの相対順序は、常にソースパーティションとは異なる場合があります。

サービスによって割り当てられたメタデータ

ソースイベントハブから取得されたイベントのサービスによって割り当てられたメタデータ、元のエンキュー時刻、シーケンス番号、およびオフセットは、ターゲットイベントハブのサービスによって割り当てられた新しい値に置き換えられますが、ヘルパー関数 (サンプルで提供されているレプリケーションタスク) では、元の値は次のユーザープロパティに保持されます: repl-enqueue-time (ISO8601 文字列型)、repl-sequence、repl-offset。

これらのプロパティは文字列型であり、それぞれの元のプロパティの文字列化された値が含まれています。複数回イベントが転送される場合は、直接のソースのサービスによって割り当てられたメタデータが既存のプロパティに追加され、値はセミコロンで区切られます。

[フェールオーバー]

Event Hubs サービスのリージョンの可用性イベントから、あるいはネットワークの中断から保護するために、ディザスターリカバリーの目的でレプリケーションを使用している場合、そのような障害シナリオでは、あるイベントハブから次のものへのフェールオーバーを実行し、プロデューサーやコンシューマーに対して、セカンダリエンドポイントを使用するように指示する必要があります。

すべてのフェールオーバーシナリオでは、名前空間の必須要素が構造的に同一であることが前提となります。これは、Event Hubs とコンシューマーグループが同じ名前であり、Shared Access Signature 規則やロールベースのアクセス制御規則が同じ方法で設定されていることを意味します。名前空間の移動に関するガイダンスに従い、クリーンアップ手順を省略して、セカンダリ名前空間を作成 (および更新) することができます。

プロデューサーとコンシューマーを強制的に切り替えるには、参照にどの名前空間を使用するかに関する情報をアクセスと更新が容易な場所で利用可能にする必要があります。プロデューサーまたはコンシューマーは頻繁に発生するまたは永続的なエラーを検出した場合、その場所を確認し、それらの構成を調整する必要があります。その構成を共有するためのさまざまな方法がありますが、ここでは次の 2 つを取り上げます: DNS とファイル共有。

DNS ベースのフェールオーバー構成

1 つの候補として、制御する DNS で DNS SRV レコードの情報を保持し、それぞれのイベントハブエンドポイントを指す方法があります。

重要

Event Hubs では、そのエンドポイントを CNAME レコードで直接エイリアス化することが許可されないことに注意してください。つまり、DNS は、IP アドレス情報を直接解決するためではなく、エンドポイントアドレスに対する回復力のある参照メカニズムとして使用します。

ドメイン example.com を所有しているとします。アプリケーションのゾーンは test.example.com です。 2 つの代替の Event Hubs に対して、ここでさらに入れ子になった 2 つのゾーンおよびそれぞれに SRV レコードを作成します。

SRV レコードには、一般的な規則に従って、先頭に _azure_eventhubs._amqp が付けられ、2 つのエンドポイントレコードが保持されます。1 つはポート 5671 の AMQP-over-TLS 用で、もう 1 つはポート 443 の AMQP-over-WebSockets 用です。これらの両方が、ゾーンに対応する名前空間の Event Hubs エンドポイントを指します。

ゾーン	SRV レコード
`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`

次に、アプリケーションのゾーンで、プライマリイベントハブに対応する下位ゾーンを指す CNAME エントリを作成します。

CNAME レコード	エイリアス
`eventhub.test.example.com`	`eh1.test.example.com`

その後、明示的な CNAME および SRV レコードに対するクエリの実行を許可する DNS クライアントを使用して (Java および .NET の組み込みクライアントで許可されるのは、IP アドレスへの名前のシンプルな解決のみ)、目的のエンドポイントを解決できます。たとえば、DnsClient.NET の場合、参照関数は次のようになります。

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('.');
}

この関数からは、上記のように CNAME で現在エイリアス化されているゾーンのポート 5671 に登録されているターゲットホスト名が返されます。

フェールオーバーを実行するには、CNAME レコードを編集し、代替ゾーンを指定する必要があります。

DNS (具体的には Azure DNS) を使用する利点は、Azure DNS の情報がグローバルにレプリケートされるため、単一リージョンの障害に対する回復性があることです。

この手順は、Event Hubs Geo-DR の場合と似ていますが、ユーザーが自分で完全に制御でき、アクティブ/アクティブシナリオでも機能します。

エンドポイント情報の共有に DNS を使用する最もシンプルな代替方法は、プライマリエンドポイントの名前をプレーンテキストファイルに格納し、障害に対して堅牢で、引き続き更新を許可するインフラストラクチャからファイルを提供することです。

グローバル対応でコンテンツレプリケーションが可能な高可用性 Web サイトインフラストラクチャを既に実行している場合は、このようなファイルをそこに追加し、切り替えが必要な場合はファイルを再発行します。

注意

エンドポイント名は、シークレットを含む完全な接続文字列ではなく、この方法でのみ公開する必要があります。

コンシューマーのフェールオーバーに関する追加の考慮事項

イベントハブコンシューマーの場合、フェールオーバー戦略に関するその他の考慮事項は、イベントプロセッサのニーズによって異なります。

バックアップデータから、データベースを含むシステムを再構築する必要がある障害が発生し、そのデータベースが直接、またはイベントハブに保持されているイベントからの中間処理を介してフィードされる場合は、バックアップを復元してから、元のシステムが破棄された時点からではなく、データベースのバックアップが作成された時点からシステムへのイベントの再生を開始する必要があります。

障害がシステムのスライスにのみ影響する場合や、実際には到達できなくなったイベントハブが 1 つだけの場合は、処理が中断されたのとほぼ同じ位置からイベントの処理を続行するのが一般的です。

いずれかのシナリオを実現し、それぞれの Azure SDK のイベントプロセッサを使用するには、新しいチェックポイントストアを作成し、処理を再開する "タイムスタンプ" に基づいて、初期のパーティション位置を指定します。

切り替え元のイベントハブのチェックポイントストアに引き続きアクセスできる場合は、既に処理されているイベントをスキップし、最後に中断した場所から正確に再開するのに、前述の伝達されたメタデータが役立ちます。

マージする

マージパターンには、1 つのターゲットを指す 1 つまたは複数のレプリケーションタスクがあります。また、通常のプロデューサーと同時に、同じターゲットにイベントが送信される場合もあります。

これらのパターンの種類は次のとおりです。

2 つ以上のレプリケーション関数で、別々のソースからイベントを同時に取得し、それらを同じターゲットに送信する。
もう 1 つのレプリケーション関数で、ソースからイベントを取得するが、ターゲットもプロデューサーによって直接使用される。
パターンは前述のものだが、2 つ以上のイベントハブ間でミラー化されるため、イベントの生成場所に関係なく、それらのイベントハブには同じストリームが含まれる。

最初の 2 つのパターンの違いはわずかであり、単純なレプリケーションタスクと異なりません。

最後のシナリオでは、既にレプリケートされているイベントが再度レプリケートされないようにする必要があります。この手法については、EventHubToEventHubMerge サンプルで説明されています。

エディター

エディターパターンはレプリケーションパターンに基づいていますが、メッセージは転送される前に変更されます。

そのような変更の例を以下に示します。

"コード変換" - Apache Avro 形式または何らかの独自のシリアル化形式を使用してエンコードされたソースからイベントコンテンツ ("本文" または "ペイロード" ともいう) が到着したものの、ターゲットを所有しているシステムではコンテンツが JSON エンコードされることを想定している場合、コード変換レプリケーションタスクでは、まず Apache Avro からメモリ内オブジェクトグラフにペイロードを逆シリアル化してから、そのグラフを、転送されるイベントのために JSON 形式にシリアル化します。また、コード変換にはコンテンツの圧縮とその解除のタスクも含まれます。
"変換" - 構造化データを含むイベントは、ダウンストリームコンシューマーによる使用をより容易にするために、そのデータをリシェイプする必要がある場合があります。これには、入れ子構造のフラット化、余分なデータ要素の排除、特定のスキーマに正確に適合させるためのペイロードのリシェイプなどの作業が含まれる場合があります。
"バッチ処理" - ソースからバッチでイベント (1 回の転送で複数のイベント) が受信される場合がありますが、ターゲットに 1 つずつ転送する必要があります。その逆の場合も同様です。そのため、タスクで 1 つの入力イベントの転送に基づいて複数のイベントが転送されたり、一連のイベントがまとめて転送されたりする場合があります。
"検証" - 外部ソースからのイベントデータでは多くの場合、転送する前に、一連の規則に準拠しているかどうかを確認する必要があります。これらの規則はスキーマまたはコードを使用して表すことができます。準拠していないことが検出されたイベントは、ログに記録された問題と共に削除されることもあれば、特別なターゲット宛先に転送されてさらに処理されることもあります。
"エンリッチメント" - 一部のソースから送信されるイベントデータは、ターゲットシステムで使用できるようにするためにさらにコンテキストでのエンリッチメントが必要な場合があります。これには、参照データを検索し、そのデータをイベントと共に埋め込んだり、レプリケーションタスクで認識されてはいるもののイベントには含まれていないソースに関する情報を追加したりする作業が含まれる場合があります。
"フィルター処理" - ソースから到着する一部のイベントは、何らかの規則に基づいてターゲットから除外する必要がある場合があります。フィルターにより、規則に対してイベントがテストされ、その規則と一致しない場合はイベントが削除されます。特定の条件を観察し、同じ値の後続のイベントを削除することによって重複イベントをフィルター処理することは、フィルター処理の 1 つの形式です。
"暗号化" - レプリケーションタスクでは、ソースから到着するコンテンツの暗号化を解除したり、ターゲットに転送されるコンテンツを暗号化したりする必要がある場合があります。また、イベントで伝達される署名を基準にコンテンツやメタデータの整合性を確認したり、そのような署名を添付したりする必要がある場合があります。
"構成証明" - レプリケーションタスクでは、特定のチャネルを介して、あるいは特定の時刻にイベントが受信されたことを証明するイベントに、デジタル署名で保護される可能性のあるメタデータを添付する場合があります。
"チェーン" - レプリケーションタスクでは、ストリームの整合性が保護され、欠落しているイベントを検出できるように、イベントのストリームに署名を適用する場合があります。

通常、変換、バッチ処理、およびエンリッチメントのパターンは Azure Stream Analytics ジョブで実装することをお勧めします。

これらのパターンはすべて Azure Functions を使用して実装でき、Event Hubs トリガーを使用してイベントを取得し、イベントハブの出力バインドを使用してそれらを配信します。

ルーティング

ルーティングパターンはレプリケーションパターンに基づいていますが、1 つのソースと 1 つのターゲットがあるのではなく、レプリケーションタスクに複数のターゲットがあります。これについては、以下に C# で示します。

[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;
        });
    }
}

ルーティング関数では、メッセージメタデータやメッセージペイロードを考慮し、送信先として使用可能な宛先のいずれかを選択します。

Azure Stream Analytics では、複数の出力を定義し、出力ごとにクエリを実行することで、同じことを実現できます。

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

ログプロジェクション

ログプロジェクションパターンでは、イベントストリームがインデックス付きデータベースにフラット化され、イベントはデータベース内のレコードになります。通常、イベントは同じコレクションまたはテーブルに追加され、イベントハブパーティションキーは、レコードを一意にするために主キーの一部になります。

ログプロジェクションでは、イベントデータの時系列ヒストリアンまたは圧縮されたビューを生成できます。これにより、パーティションキーごとに最新のイベントのみが保持されます。ターゲットデータベースのシェイプは、最終的にはユーザーとユーザーのアプリケーションのニーズによって決まります。このパターンは "イベントソーシング" とも呼ばれます。

ヒント

Azure Stream Analytics で Azure SQL Database と Azure Cosmos DB へのログプロジェクションを簡単に作成することができます。このオプションを選択することをお勧めします。

次の Azure 関数では、Azure Cosmos DB コレクションに圧縮されたイベントハブの内容を射影します。

[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);
        }
    }
}

次の方法で共有

イベントレプリケーションタスクのパターン

レプリケーション

ストリームと順序の維持

サービスによって割り当てられたメタデータ

[フェールオーバー]

DNS ベースのフェールオーバー構成

コンシューマーのフェールオーバーに関する追加の考慮事項

マージする

エディター

ルーティング

ログプロジェクション

次のステップ

フィードバック

その他のリソース

次の方法で共有

イベント レプリケーション タスクのパターン

レプリケーション

ストリームと順序の維持

サービスによって割り当てられたメタデータ

[フェールオーバー]

DNS ベースのフェールオーバー構成

ファイル共有ベースのフェールオーバー構成

コンシューマーのフェールオーバーに関する追加の考慮事項

マージする

エディター

ルーティング

ログ プロジェクション

次のステップ

フィードバック

その他のリソース

イベントレプリケーションタスクのパターン

ログプロジェクション