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

##やりたいこと
Event Hubsでデータを受信し、Stream AnalyticsからCosmos DBにデータ投入するアーキテクチャを構築します。
構築後は、PostmanからEvent Hubsにデータ投入して動作確認をします。
![構成.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/d0dad275-67c9-b190-3f73-c20c4a6ed722.png)
ただ投げ込むだけではつまらないので、「リアルタイムの不正検出」というシナリオにしました。
クライアントから簡単なjsonファイルを受け取り、そのデータ内容に従ってStream Analyticsで分岐処理し、それぞれCosmos DBのOKデータコンテナ、NGデータコンテナに振り分けます。
![データの流れ.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/8fc7c8fd-5a58-8204-8ad4-a3070d8c29ad.png)

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

##１．Event Hubsの作成
参考：[クイック スタート:Azure portal を使用したイベント ハブの作成](https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-create)
####１．イベントハブ名前空間の作成
* 名前空間の価格レベルはBasicで作成します。
![名前空間の作成.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/965f239a-3a16-859f-cc7a-3698a6c597fe.png)

####２．イベントハブの作成
* 名前空間の中に、１つイベントハブを作成します。
![イベントハブ作成.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/eb3c4c51-e22c-0522-ef23-a6fbf8765f89.png)


##２．Cosmos DBの作成
参考：[クイック スタート:Azure portal を使用して Azure Cosmos のアカウント、データベース、コンテナー、および項目を作成する](https://docs.microsoft.com/ja-jp/azure/cosmos-db/create-cosmosdb-resources-portal)
####１．Cosmos DB アカウントの作成
* APIは必ず「コア(SQL)」で作成します。（そのほかのAPIはStream AnalyticsからのINPUTを保証していないため。）
![cosmosdbの作成.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/fa99e7ae-d047-aefd-b7fb-356f38c52238.png)

####２．コンテナの作成
* コンテナは「OK用データ格納コンテナ」と「NG用データ格納コンテナ」の２つを作成します。
* コンテナIDは「ok_items」と「ng_items」で、パーティションキーはどちらも「/category」です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/412a4612-45d1-5766-3280-d6e72dd8fda6.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/25b01cc5-718c-e86c-cb43-9ac1062cc9c5.png)


##３．Stream Analyticsの作成
参考：[Azure Stream Analytics を使用してイベント ハブからのデータを処理する](https://docs.microsoft.com/ja-jp/azure/event-hubs/process-data-azure-stream-analytics)

* ホスティング環境は「クラウド」です。
![StreamAnalytics作成.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/511b54ee-1878-f9f3-9948-d51c4434447c.png)


##４．Stream Analyticsのクエリ作成
####１．入力の設定

* ジョブ トポロジ -> 入力 -> ストリーム入力の追加 -> **イベントハブ**を選択

* 入力のエイリアス：任意の値を入力。この値がクエリ内で変数として使用できます。
![入力.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/d0add529-4291-41e2-e9b2-f7414ba0263c.png)

####２．出力の設定
* ジョブ トポロジ -> 出力 -> 追加 -> **Cosmos DB**を選択
* 出力先は、OKコンテナとNGコンテナの２つ。
* 出力のエイリアス：任意の値を入力。この値がクエリ内で変数として使用できる。
![出力１.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/b1e612eb-0fa5-8ac2-bcdd-28211847e803.png)

####３．クエリの作成
* 下記SQLをクエリに入力する。

```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'
```

* ポータル上では下記のように見える。
* クエリ内の変数は、**入力・出力エイリアス名**となっている。
![クエリ作成.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/e37f5a3a-7314-6506-58b7-83168a09a424.png)



##５．SASトークンの取得
参考：
[Azure EventhubsにSASトークンを使う方法(Java)](https://qiita.com/kenseitogari1/items/1f6b8d956a3ad7a54b93)
[Azure Event Hubs に REST API でログを送りつける話](https://devlog.arksystems.co.jp/2016/06/29/683/)
[Generate SAS token](https://docs.microsoft.com/en-us/rest/api/eventhub/Generate-SAS-token?redirectedfrom=MSDN)

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

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

　①共有アクセスポリシー名
　②主キー
　③接続文字列　主キー
　①～③を取得し、**④はすべてにチェックを付けます**
![共有アクセスポリシーの取得.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/bd642cd5-08e5-bb75-f661-98e3a990ac45.png)

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

```javascript: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トークン　<br> Content-Type：application/json|
| ③| Bodyタブ | Jsonデータ |
![postman１.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/14a7aca9-a2f2-82e3-01af-1529064148c0.png)
![postman２.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/7be55ead-50f5-1e87-c3c8-ad6816f54675.png)

```json:Bodyデータ
{
    "id": "1",
    "category": "ok",
    "name": "ok_request",
    "description": "test data 1"
}
```

####２．動作確認
* Stream Analyticsを開始してから、Postman実行します。
![開始.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/2157d338-fd84-30e3-4773-e968984579b1.png)

* 201が返ってきたら成功！
![成功.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/f3e88c58-35a9-74dd-4cf3-939c01a97c22.png)

* ポータルから、各サービスを見ていきます。

　**Event Hubs**
　メッセージの受信が確認できます。
　Incomingが1、Outgoingが2になっているのは、おそらくStream AnalyticsのクエリでEvent Hubsからのメッセージを
　2回呼んでいる（FROM句が2回入っている）からです。
![EventHubs_OK.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/22232dd5-1f48-a078-30cf-a8e2b36c55ba.png)


　**Stream Analytics**
　こちらも、イベントの受信が確認できます。
　クエリでEvent Hubsからのイベントを2回呼び、Cosmos DBに対して1つのデータを受け渡しています。
![StreamAnalytics_OK.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/14964947-60f7-22c8-76df-24382a1a6cee.png)


　**Cosmos DB**
　OKデータ格納コンテナに、Postmanから投入したデータがちゃんと入っています。
　「EventProcessedUtcTime」以降の項目は、Stream Analyticsからデータが渡されると自動で付与されるものです。
![CosmoDB_OK.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/4865d3d1-2d7f-1d36-40de-55188dd34a57.png)


　NGデータを投入すると、ちゃんとNGデータ格納コンテナにデータが入ってきました。
![CosmoDB_NG.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/453255/5a380a3e-372f-fd32-24bc-ade577947cd3.png)



##７．ハマったところ
####１．SASトークンの取得
Event Hubsに対するSASトークンの取得で詰まりました。
今回Event Hubsはベーシックプランで作成したのですが、スタンダードプランとベーシックプランでSASトークンの取得方法が変わるようです。この記事では、ベーシックプラン用のSASトークン取得スクリプトを載せています。
####２．Postmanで401が返ってくる。
Event Hubsの共有アクセスポリシーの設定が足りていませんでした。
共有アクセスポリシーは、管理・送信・リッスンのすべてにチェックを付けてください。


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