EventHubs→Stream Analytics→CosmosDBにデータ投入する #Azure

やりたいこと

Event Hubsでデータを受信し、Stream AnalyticsからCosmos DBにデータ投入するアーキテクチャを構築します。
構築後は、PostmanからEvent Hubsにデータ投入して動作確認をします。

ただ投げ込むだけではつまらないので、「リアルタイムの不正検出」というシナリオにしました。
クライアントから簡単なjsonファイルを受け取り、そのデータ内容に従ってStream Analyticsで分岐処理し、それぞれCosmos DBのOKデータコンテナ、NGデータコンテナに振り分けます。

手順

１．Event Hubsの作成
２．Cosmos DBの作成
３．Stream Analyticsの作成
４．Stream Analyticsのクエリ作成
５．SASトークンの取得
６．Postmanでのテスト
７．ハマったところ
８．感想

１．Event Hubsの作成

参考：クイックスタート:Azure portal を使用したイベントハブの作成

１．イベントハブ名前空間の作成

名前空間の価格レベルはBasicで作成します。

２．イベントハブの作成

名前空間の中に、１つイベントハブを作成します。

２．Cosmos DBの作成

参考：クイックスタート:Azure portal を使用して Azure Cosmos のアカウント、データベース、コンテナー、および項目を作成する

１．Cosmos DB アカウントの作成

APIは必ず「コア(SQL)」で作成します。（そのほかのAPIはStream AnalyticsからのINPUTを保証していないため。）

２．コンテナの作成

コンテナは「OK用データ格納コンテナ」と「NG用データ格納コンテナ」の２つを作成します。
コンテナIDは「ok_items」と「ng_items」で、パーティションキーはどちらも「/category」です。

３．Stream Analyticsの作成

参考：Azure Stream Analytics を使用してイベントハブからのデータを処理する

ホスティング環境は「クラウド」です。

４．Stream Analyticsのクエリ作成

１．入力の設定

ジョブトポロジ -> 入力 -> ストリーム入力の追加 -> イベントハブを選択
入力のエイリアス：任意の値を入力。この値がクエリ内で変数として使用できます。

２．出力の設定

ジョブトポロジ -> 出力 -> 追加 -> Cosmos DBを選択
出力先は、OKコンテナとNGコンテナの２つ。
出力のエイリアス：任意の値を入力。この値がクエリ内で変数として使用できる。

３．クエリの作成

下記SQLをクエリに入力する。

クエリ


SELECT
    *
INTO
    [cosmosdb-ok-request]
FROM
    [eventhubs-request]
WHERE
    [eventhubs-request].category = 'ok'
SELECT
    *
INTO
    [cosmosdb-ng-request]
FROM
    [eventhubs-request]
WHERE
    [eventhubs-request].category = 'ng'

ポータル上では下記のように見える。
クエリ内の変数は、入力・出力エイリアス名となっている。

５．SASトークンの取得

参考：
Azure EventhubsにSASトークンを使う方法(Java)
Azure Event Hubs に REST API でログを送りつける話
 Generate SAS token

１．Event Hubs ベーシック用のSASトークンをJavascriptで作成

イベントハブ名前空間 -> イベントハブ -> 共有アクセスポリシー

　①共有アクセスポリシー名
　②主キー
　③接続文字列　主キー
　①～③を取得し、④はすべてにチェックを付けます

２．上記の情報をもって、SASトークン取得スクリプトを作成する。

create_sas.js

var crypto = require('crypto');
var saName = "①共有アクセスポリシー名";
var saKey = "②主キー";
var uri = "③sb://イベントハブ名前空間.servicebus.windows.net/イベントハブ名";

createSharedAccessToken(uri, saName, saKey);

function createSharedAccessToken(uri, saName, saKey) { 
    if (!uri || !saName || !saKey) { 
            throw "Missing required parameter"; 
        } 
    var encoded = encodeURIComponent(uri); 
    var now = new Date(); 
    var week = 60*60*24*7;
    var ttl = Math.round(now.getTime() / 1000) + week;
    var signature = encoded + '\n' + ttl; 
    var signatureUTF8 = encode_utf8(signature); 
    var hash = crypto.createHmac('sha256', saKey).update(signatureUTF8).digest('base64'); 
    console.log('SharedAccessSignature sr=' + encoded + '&sig=' +  
    encodeURIComponent(hash) + '&se=' + ttl + '&skn=' + saName);
}

function encode_utf8(s) {
    return unescape(encodeURIComponent(s));
}

３．作成したスクリプトで、SASトークンを取得する

C:\Users>node create_sas.js
SharedAccessSignature sr=sb%3A%2F%2F ~中略~ _policy

６．Postmanでのテスト

４までで作成したサービスに対して、５で作成したSASトークンを使ってPostmanで動作確認

１．Postmanの準備

	項目	値
①	URL	https://名前空間.servicebus.windows.net/イベントハブ名/messages
②	Headersタブ	Authorization：５で取得したSASトークン　 Content-Type：application/json
③	Bodyタブ	Jsonデータ

Bodyデータ

{
    "id": "1",
    "category": "ok",
    "name": "ok_request",
    "description": "test data 1"
}

２．動作確認

Stream Analyticsを開始してから、Postman実行します。
201が返ってきたら成功！
ポータルから、各サービスを見ていきます。

　Event Hubs
　メッセージの受信が確認できます。
　Incomingが1、Outgoingが2になっているのは、おそらくStream AnalyticsのクエリでEvent Hubsからのメッセージを
　2回呼んでいる（FROM句が2回入っている）からです。

　Stream Analytics
　こちらも、イベントの受信が確認できます。
　クエリでEvent Hubsからのイベントを2回呼び、Cosmos DBに対して1つのデータを受け渡しています。

　Cosmos DB
　OKデータ格納コンテナに、Postmanから投入したデータがちゃんと入っています。
　「EventProcessedUtcTime」以降の項目は、Stream Analyticsからデータが渡されると自動で付与されるものです。

　NGデータを投入すると、ちゃんとNGデータ格納コンテナにデータが入ってきました。

７．ハマったところ

１．SASトークンの取得

Event Hubsに対するSASトークンの取得で詰まりました。
今回Event Hubsはベーシックプランで作成したのですが、スタンダードプランとベーシックプランでSASトークンの取得方法が変わるようです。この記事では、ベーシックプラン用のSASトークン取得スクリプトを載せています。

２．Postmanで401が返ってくる。

Event Hubsの共有アクセスポリシーの設定が足りていませんでした。
共有アクセスポリシーは、管理・送信・リッスンのすべてにチェックを付けてください。

８．感想

ポータル上ですべて構築完了するので、とても簡単です。
ハマったところは全部、サービス構築外の部分です。導入コストが低い！
また、今回はStream Analyticsのクエリは簡素なものにしましたが、次はもうちょっと複雑な条件でやってみたいです。

EventHubs→Stream Analytics→CosmosDBにデータ投入する