Azure Synapse Link for Azure Cosmos DB API for MongoDB を試す

# はじめに

この記事は、[GitHub][1] 上で公開されている Azure Synapse Link for Azure Cosmos DB API for MongoDB のサンプルの解説記事です。
本家の記載は英語となっています。[こちら][2] のリポジトリにて日本語訳を行っていますので、どちらか好きな方にアクセスを行ってください。

- 本家 (English): [Azure-Samples/Synapse][1]
- 日本語訳 (非公式): [ymasaoka/Synapse][2]

なお、Azure Synapse Link for Azure Cosmos DB API for MongoDB のサンプルは、[/Notebooks/PySpark/Synapse Link for Cosmos DB samples/MongoDB][3] に用意されています。詳細を確認したい方は、こちらも合わせてご確認ください。


# 環境構築

このサンプルの実行には、以下の環境が必要です。

- Azure Cosmos DB アカウント (API for MongoDB)
- Spark プールがある Azure Synapse ワークスペース

環境の作成方法については、以下の記事を参考にしてみてください。
Cosmos DB の要求ユニット (RU/s) は 400 RU/s で十分です。

- [Azure Synapse Link を試してみる][4]

ただし、`Azure Cosmos DB アカウント`の作成時に選択する API は、必ず **MongoDB 用 Azure Cosmos DB API** を選択してください。

![スクリーンショット 2020-09-23 14.34.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/b2f89393-4650-f2dc-b8cb-00a36cd6aa4b.png)


# Azure Cosmos DB の API for MongoDB と Synapse Link をはじめる

## 今回の流れ

今回は、以下の流れで Azure Synapse Link の動作を確認していきます。

1. 従来の MongoDB クライアントを使用してデータセットを Azure Cosmos DB トランザクションストアに挿入
2. トランザクションストアから ETL された分析ストア内のデータに対して集計クエリを実行
3. 別のデータセットを Azure Cosmos DB トランザクションストアに挿入
4. 分析ストア内にある両方のデータセットを統合し、集計クエリを実行

## 事前準備

## Cosmos DB コレクションの作成

Azure Cosmos DB 側で、データベースとコレクションを作成しましょう。
1 点注意は、MongoDB の場合、パーティションキーは シャードキー[^1]というものに置き換えられます。
また、2020/09/25 時点で、本家のサンプル記載ではシャードキーを設定するとありますが、**シャードキーは設定しないでください**。

- **データベース**: `test`
- **コレクション**: `htap`
    - **Storage capacity**: `Fixed (10 GB)`
    - **分析ストア**: `On`

![スクリーンショット 2020-09-24 1.32.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/c7e9c626-4647-6b1a-8cdb-a2d723819a01.png)

![スクリーンショット 2020-09-25 20.00.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/ea980915-df29-bc9d-c107-827fbc9905f0.png)


## Synapse 側で Linked services を作成

こちらは、SQL API の時と同様に、Azure Synapse ワークスペースより Linked services を設定します。
リンクされたサービス名の名前は任意のものを設定してください。
この記事では、`CosmosDBMongoTest`という名前を使用します。

![スクリーンショット 2020-09-24 1.50.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/80fba945-583e-4731-630c-2931aeb3c4eb.png)

![スクリーンショット 2020-09-24 1.52.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/70d19f36-6532-b247-5960-194b82001e58.png)

## 新しい Spark プールを作成

このサンプルでは Spark プールを使用して PySpark で Cosmos DB の分析ストアにアクセスします。
サンプルの実行にあたり、いくつかの Python ライブラリを Spark プール内にインストールしますが、既に別の用途で Spark プールを作成している場合、`新しい Spark プールを作成することを推奨します`。
Spark プールに Python ライブラリをインストールすることで、他に影響を与えないためです。
使用していない Spark プールについては、アイドル期間中はコンピューティングの課金は発生しません。(自動一時停止の設定)
デフォルトでは **15 分**になっていますが、こちらを無効化しないようにしておくのがベストです。アイドル状態までの時間は任意で変更してください。

![スクリーンショット 2020-09-24 2.00.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/bb8986e2-715f-990f-7c2c-8a9bfef094b3.png)

![スクリーンショット 2020-09-24 2.01.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/d827da96-12d3-8693-e245-ab62c2e1af01.png)

## サンプル実行に必要なライブラリをインストール

新しく作成した Spark プールに対して、サンプル実行に必要な Python ライブラリをインストールします。
サンプル実行に必要なライブラリの一覧は、サンプルノートブックに記載のある通りです。

```text
pymongo==2.8.1
aenum==2.1.2
backports-abc==0.5
bson==0.5.10
```

これを Spark プールにインストールします。インストールは pip 経由ではなく、`pip freeze`のリファレンスドキュメントで説明されている形式に従った**要件ファイル**を使用し、Spark プールの作成時、あるいは Azure ポータル、Synapse ワークスペースの専用画面から行います。

まず、ローカルに、`requirements.txt`を用意します。

```text:requirements.txt
pymongo==2.8.1
aenum==2.1.2
backports-abc==0.5
bson==0.5.10
```

今回は、Synapse ワークスペースの画面から、この requirements.txt を使用して Python モジュールのインストールを行います。
[Manage] -> [Apache Spark pools] 画面に遷移し、サンプルで使用する Spark プールを選択します。

![スクリーンショット 2020-09-24 3.20.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/d1f60524-e9c7-d48b-d102-4a23e2a7d162.png)

Spark プールを選択すると、プロパティ画面が表示されますので、[Packages] 欄にある **Upload environment config file** ボタンを選択し、先ほど作成した requirements.txt をアップロードしましょう。

![スクリーンショット 2020-09-24 3.22.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/6db40ff9-b9c0-0046-5877-59ca859ae251.png)

アップロードが正常に完了すると、以下のような通知が確認でき、requirements.txt が適用されたことが確認できるはずです。

![スクリーンショット 2020-09-24 3.25.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/ab621bc3-37c1-f666-7dc1-25284e04f3cc.png)

![スクリーンショット 2020-09-24 3.25.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/c6ea168b-fc95-1dd8-6b1e-ad5e70183b8c.png)

インストールされた Python ライブラリは、以下のコードで確認が可能です。Synapse ワークスペースのノートブックから実行することで確認することができます。

```python
import pip # pip 関数を使用するために必要です。
for i in pip.get_installed_distributions(local_only=True):
    print(i)
```

実行する際は、実行に使用する Spark プール、および Language が PySpark に設定されていることを確認しましょう。

![スクリーンショット 2020-09-24 3.32.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/2c5668de-6644-b99e-9c53-3b59f4d3639e.png)

requirements.txt に記載したパッケージは、Spark プールの開始時に PyPI からダウンロードされる仕組みになっており、Spark プールから Spark インスタンスが作成される度に使用されるようになります。
もし、出力結果に requirements.txt に記載した Python ライブラリが表示されていない場合は、Spark プールを再起動してみてください。
なお、requirements.txt のアップロード後は、Spark プールの再起動が内部的に実行されるため、環境によっては反映までに時間を要する場合があります。

## MongoDB クライアントを初期化しデータを挿入

Python 用の MongoDB クライアントである **pymongo** を使用して、Cosmos DB のトランザクションストアにデータを挿入します。
MongoDB クライアントを使用するにあたって、CosmosDB の`ユーザー名`と`プライマリ/セカンダリ パスワード`を先に取得しておきます。

![スクリーンショット 2020-09-25 20.26.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/348116/3af8c181-5ee8-034b-bd41-dcbe5addb42d.png)

取得したら、変数として設定しておきます。

```python
DATABASE_ACCOUNT_NAME = '<ここにユーザー名を入力>'
DATABASE_ACCOUNT_READWRITE_KEY = '<ここにプライマリ/セカンダリ パスワードを入力>'
```

設定したら、MongoDB クライアントを初期化します。

```python
from pymongo import MongoClient
from bson import ObjectId # ObjectId が機能するため

client = MongoClient("mongodb://{account}.mongo.cosmos.azure.com:10255/?ssl=true&replicaSet=globaldb".format(account = DATABASE_ACCOUNT_NAME)) # 独自のデータベースアカウントのエンドポイント。
db = client.test    # データベースを選択
db.authenticate(name=DATABASE_ACCOUNT_NAME,password=DATABASE_ACCOUNT_READWRITE_KEY) # データベースアカウント名と任意の読み取り/書き込みキーを使用します。
```

MongoDB クライアントを初期化したら、データを挿入します。500 アイテムを生成します。

```python
from random import randint
import time

orders = db["htap"]

items = ['Pizza','Sandwich','Soup', 'Salad', 'Tacos']
prices = [2.99, 3.49, 5.49, 12.99, 54.49]

for x in range(1, 501):
    order = {
        'item' : items[randint(0, (len(items)-1))],
        'price' : prices[randint(0, (len(prices)-1))],
        'rating' : randint(1, 5),
        'timestamp' : time.time()
    }
    
    result=orders.insert(order)

print('500 個の注文の作成が終了しました')
```

## 分析ストアからデータを確認

トランザクションストアにデータを挿入したデータが、分析ストアから確認できることを確認します。
データが裏側で自動的に ETL されるため、データの挿入後すぐに以下のコードを実行するとエラーになります。
5 分程度、時間をおいて実行してください。

```python
# 分析ストアのデータをデータフレームにロードする
# シークレットを使用してセルを実行し、DATABASE_ACCOUNT_NAME および DATABASE_ACCOUNT_READWRITE_KEY 変数を取得します。
df = spark.read.format("cosmos.olap")\
    .option("spark.cosmos.accountEndpoint", "https://{account}.documents.azure.com:443/".format(account = DATABASE_ACCOUNT_NAME))\
    .option("spark.cosmos.accountKey", DATABASE_ACCOUNT_READWRITE_KEY)\
    .option("spark.cosmos.database", "test")\
    .option("spark.cosmos.container", "htap")\
    .load()

# ピザの注文からのすべての収益を調べてみましょう
df.groupBy(df.item.string).sum().show()

# df[df.item.string == 'Pizza'].show(10) 
# df.select(df['item'] == Struct).show(10) 
# df.select("timestamp.float64").show(10)
# df.select("timestamp.string", when(df.timestamp.string != null)).show(10)
```

結果が返ってきたら、一緒にスキーマ情報も確認してみましょう。

```python
df.schema
```

```:実行結果(見やすいように改行しています)
StructType(List(StructField(_rid,StringType,true),
StructField(_ts,LongType,true),StructField(id,StringType,true),
StructField(_etag,StringType,true),
StructField(_id,StructType(List(StructField(objectId,StringType,true))),true),
StructField(item,StructType(List(StructField(string,StringType,true))),true),
StructField(price,StructType(List(StructField(float64,DoubleType,true))),true),
StructField(rating,StructType(List(StructField(int32,IntegerType,true))),true),
StructField(timestamp,StructType(List(StructField(float64,DoubleType,true))),true),
StructField(_partitionKey,StructType(List(StructField(string,StringType,true))),true)))
```

## 形式が異なる timestamp データを追加で投入

さらにデータを投入していきます。
追加のデータ投入では、**timestamp**データの部分は、先ほどは`time.time()`を指定していましたが、今回は`strftime("%Y-%m-%d %H:%M:%S")`を指定してみます。

```python
from random import randint
from time import strftime

orders = db["htap"]

items = ['Pizza','Sandwich','Soup', 'Salad', 'Tacos']
prices = [2.99, 3.49, 5.49, 12.99, 54.49]

for x in range(1, 501):
    order = {
        'item' : items[randint(0, (len(items)-1))],
        'price' : prices[randint(0, (len(prices)-1))],
        'rating' : randint(1, 5),
        'timestamp' : strftime("%Y-%m-%d %H:%M:%S")
    }
    
    result=orders.insert(order)

print('500 個の注文の作成が終了しました')
```

もし、上記のコードの実行に時間がかかっている、またはエラーが返された場合、MongoDB クライアントをもう一度初期化した後、再実行してみてください。

## 追加したデータを確認

データの投入が完了したら、再度、分析ストアのデータを確認しましょう。
前述の通り、トランザクションストアから分析ストアに対してデータがバックグラウンドで ETL されているため、少し時間をおいて実行してください。

```python
# 分析ストアのデータをデータフレームにロードする
# シークレットを使用してセルを実行し、DATABASE_ACCOUNT_NAME および DATABASE_ACCOUNT_READWRITE_KEY 変数を取得します。
df = spark.read.format("cosmos.olap")\
    .option("spark.cosmos.accountEndpoint", "https://{account}.documents.azure.com:443/".format(account = DATABASE_ACCOUNT_NAME))\
    .option("spark.cosmos.accountKey", DATABASE_ACCOUNT_READWRITE_KEY)\
    .option("spark.cosmos.database", "test")\
    .option("spark.cosmos.container", "htap")\
    .load()

# ピザの注文からのすべての収益を調べてみましょう
df.filter( (df.timestamp.string != "")).show(10)
```

ここでのポイントは、timestamp.string パラメーターを指定していることです。これを指定することによって、time.time() で挿入されたデータではなく、strftime("%Y-%m-%d %H:%M:%S") で挿入された、ISO 文字列の日付が入っているデータのみを個別に読み取ることができます。
合わせて、先ほど同様、スキーマ情報も確認してみましょう。

```python
df.schema
```

```:実行結果(見やすいように改行しています)
StructType(List(StructField(_rid,StringType,true),
StructField(_ts,LongType,true),
StructField(id,StringType,true),
StructField(_etag,StringType,true),
StructField(_id,StructType(List(StructField(objectId,StringType,true))),true),
StructField(item,StructType(List(StructField(string,StringType,true))),true),
StructField(price,StructType(List(StructField(float64,DoubleType,true))),true),
StructField(rating,StructType(List(StructField(int32,IntegerType,true))),true),
StructField(timestamp,StructType(List(StructField(float64,DoubleType,true),StructField(string,StringType,true))),true),
StructField(_partitionKey,StructType(List(StructField(string,StringType,true))),true)))
```

timestamp の部分が`StructField(timestamp,StructType(List(StructField(float64,DoubleType,true),StructField(string,StringType,true))),true),`に変更されていることを確認できると思います。

これによって、Azure Cosmos DB 分析ストアのスキーマ管理にて、スキーマ情報が変更されたことを確認することができると思います。

---
[^1]: MongoDB のシャーディングについては、(古い書籍ですが)[スケーリングMongoDB](https://www.oreilly.co.jp/books/9784873115221/) を読むと良いかと思います。※MongoDB 1.x 当時の本です。(現在の最新バージョンは 4.4)

[1]:https://github.com/Azure-Samples/Synapse
[2]:https://github.com/ymasaoka/Synapse
[3]:https://github.com/ymasaoka/Synapse/tree/ja-jp/Notebooks/PySpark/Synapse%20Link%20for%20Cosmos%20DB%20samples/MongoDB
[4]:https://qiita.com/ymasaoka/items/3caac5045f0e5349504f