Azure Event Hubs で送受信

あるアプリケーションから別のアプリケーションへデータを送信する方法として、
Microsoft Azure のイベント ハブ (Event Hubs) を利用できます。
イベント ハブは、小さいデータを短い間隔で送るというシナリオに向いています。
また、イベント ハブは文字通りハブとして機能し、送信側と受信側を多対多とすることができます。

今回は、基本的には公式のチュートリアルである Event Hubs の使用の方法をもとにして実装していきます。
(そのため、重複する内容は割愛することがあります。)

まず、Azure の管理ポータルでイベント ハブとストレージを作成します。
ストレージは、受信のための構成を保存しておく場所になるようです。

サービス バスの名前空間を sakapon-event-201508-ns、イベント ハブの名前を sakapon-event-201508 としました。

イベント ハブ

共有アクセス ポリシーには、SendRule と ReceiveRule を定義しておきます。

イベント ハブ

 

続いて、Visual Studio でプロジェクトを作成します。
今回は、送信側も受信側も WPF アプリケーションとして作成しました。

プロジェクトを作成したら、NuGet で、送信側には WindowsAzure.ServiceBus を、
受信側には Microsoft.Azure.ServiceBus.EventProcessorHost をインストールします。
このとき、App.config には接続文字列などの設定のためのテンプレートが追加されます。

以下、送信側および受信側の実装について説明します。

 

送信側の実装

まず、App.config の appSettings に、キーが Microsoft.ServiceBus.ConnectionString のエントリが生成されているので、
そこに SendRule に対応する接続文字列を指定します。

App.config

EventHubClient オブジェクトを初期化するときに、
サービス バスの名前空間ではなく、イベント ハブの名前を指定します。
特殊な指定方法ですが、これで接続文字列が .config ファイルから読み込まれます。

あとは、送りたいメッセージをバイナリ データ (byte[] や Stream) に変換して、
Send または SendAsync メソッドに渡すだけです。
複数の属性値を持つオブジェクトを送信するには、JSON などにシリアライズするとよいでしょう。

 

受信側の実装

送信側のときと同様、App.config に、ReceiveRule に対応する接続文字列を設定します。
なお、イベント ハブのほかに、ストレージの接続文字列も設定します。

EventProcessorHost オブジェクトを初期化するには接続文字列が必要になりますが、
送信側のときとは異なり、.config ファイルから透過的に読み込む機能はないため、明示的に接続文字列を取得しています。

App.config

通知されたイベントを実際に処理するクラスとして、IEventProcessor インターフェイスを実装したクラスを作成しなければなりません。
パーティション 1 つにつき、そのインスタンスが 1 つ生成されます。

ここでは StaticEventProcessor クラスを作成しましたが、
内部的に既定のコンストラクターが呼び出されて初期化されるため、外からデリゲートを渡すのが難しい構造です。
実装がややこしくなるため、この SDK には改良の余地があるように思えます。

 

さて、上記の実装をもとに、送信側の WPF アプリケーション (SenderWpf) をマウスなどでドラッグして移動させると
受信側の WPF アプリケーション (ReceiverWpf) が自動的に同じ座標に追従するというサンプルを作成しました。
完全なソースコードは EventHubsSample – GitHub にあります。

Event Hubs - sync apps

なお、その他の注意点は以下の通りです。

  • イベントの到着の順序は保証されません。
    (上のサンプルで非連続な動作があるのはそのためです。)
  • PartitionContext や EventData は一意の SequenceNumber を持ちますが、
    これらはパーティションの中で一意であって、コンシューマー グループ全体で一意になるわけではありません。
    したがって、メッセージに一意性を持たせるには、ID や時刻を含めて送信する必要があります。
  • それぞれのデータは、コンシューマー グループ 1 つにつき 1 回だけ配信されます。
    同じコンシューマー グループに設定したアプリケーションを 2 つ起動すると、
    受信するデータが半分ずつ程度に分散してしまいます。
    コンシューマー グループを追加することにより、送信側と受信側を多対多にすることができます。
  • PartitionContext.CheckpointAsync メソッドにより、イベントを受信したことを記録します。
    これが呼び出されないと、また同じイベントが通知されます。
    また、送信側が稼働しているのに受信側が稼働していない場合もデータは蓄積されているため、
    次に受信側を起動したときに大量に押し寄せてくることがあります。

 

作成したサンプル
EventHubsSample (GitHub)

バージョン情報
.NET Framework 4.5
WindowsAzure.ServiceBus 3.0.1
Microsoft.Azure.ServiceBus.EventProcessorHost 2.0.2

参照
Event Hubs の使用

カテゴリー: クラウド. タグ: , . Leave a Comment »

クラスタリングを実装する (C#)

以前に Azure Machine Learning で色のクラスタリングを投稿しましたが、
このときはクラウド サービスである Azure Machine Learning を利用してクラスタリングを実行させていました。
しかし、この方法ではネットワークが必要であり、少なくとも現在提供されている機能では、
クライアント アプリケーションなどでリアルタイムに再学習させていくのは難しそうです。

そこで今回は、k 平均法 (k-means 法) によるクラスタリングを自前で実装してみます。

クラスタリングの対象となる各データはそれぞれの特徴を表す属性の集合を持ちますが、
k 平均法は、この属性の集合をベクトル (次元数は属性数) と見なし、
それらの距離が近いもの同士を同じクラスターに振り分けるという単純なものです。

k 平均法の詳細の手順は次の通りです。
なお、各クラスターにおけるデータの平均値を表すベクトルを重心 (centroid) と呼びます。

(0) クラスター数および反復回数を与える

(1) データの集合の中からクラスターの数だけデータを選び、各クラスターの初期の重心とする

(2) すべてのデータを、重心との距離が最も小さいクラスターに割り当てる

(3) (2) の結果として得られた各クラスターで、重心を再計算する

(4) (2)(3) を与えられた回数だけ反復する

この k 平均法のアルゴリズムについては、K-means 法を D3.js でビジュアライズしてみたを参照するとよいでしょう。
クラスタリングの処理が収束していく様子が視覚化されており、理解しやすいと思います。

さて、このアルゴリズムを C# で次のように実装してみました。
(ただし、ここに載せているのは一部です。完全なものは ClusteringConsole – GitHub にあります。)

対象のデータとして System.Drawing.Color で定義されている色の集合を利用し、属性は R, G, B の 3 つとしました。
このコンソール アプリケーションを実行すると、次の図のように結果が出力されます (結果は毎回異なります)。

ClusteringConsole

 

とりあえず単純な方法で実装してみましたが、この方法では、
(1) の段階で等しい値のデータが別々のクラスターの重心として割り当てられると、
データを持たないクラスターが出てくる可能性があります。
したがって、初期の重心を分散させるなど、改良の余地はありそうです。

ちなみに今回はコンソール アプリケーションで作成しましたが、
クラスタリングの結果を視覚化すれば Azure Machine Learning で色のクラスタリング (2) の結果と同じようになるでしょう:

Azure Machine Learning で色のクラスタリング

 

作成したサンプル
ClusteringConsole (GitHub)

参照
k平均法 – Wikipedia
K-means 法を D3.js でビジュアライズしてみた
Azure Machine Learning で色のクラスタリング (1)
「クラウドではじめる機械学習 Azure ML でらくらく体験」書評