watsonx.dataにMongoDBを組み込んでPrestoエンジンでアクセスしてみる

watsonx.dataはオープンなデータレイクハウス・アーキテクチャに基づいて構築されており、大容量のデータを安価に保存できるオブジェクトストレージにテーブルを作成したり、他のデータベースを統合することができ、それらのデータベースのデータをwatsonx.dataに組み込まれているオープンソースの[Presto](https://prestodb.io/)エンジンを使って横断的にSQL文でアクセスできます。

JSONデータを管理する[MongoDB](https://www.ibm.com/jp-ja/topics/mongodb)も組み込むことができるが、SQL文でどのようにJSONデータが取得できるか気になったので調べてみた記事です。

結果として問題が発生したので、オープンソースのPrestoをローカルにインストールしてMongoDBを組み込んで確認してみました。

## この記事の内容
- 以下の環境を作って試した
  - [IBM Cloud Pak for Data](https://www.ibm.com/docs/en/cloud-paks/cp-data/4.7.x?topic=overview)(CP4D)上にwatsonx.dataを導入
    - CP4D バージョン: 4.7.3
    - watsonx.data (SW版) バージョン: 1.0.3
  - Databases for MongoDB (SaaS) バージョン: 5.0
- IBM CloudのSaaSのサービスでMongoDBを作成してwatsonx.dataに組み込んでみたら問題が発生した。チケットをあげて確認したらwatsonx.dataの既知の問題とのこと。問題はそのうちに直ると思うので、とりあえず今回はオープンソースのPrestoを構築し、MongoDBのJSONデータがどのようにアクセスできるか確認してみた。
- Pythonのプログラムでアクセスしてみたところ、JSONデータは階層化された配列で受信できることがわかった。さらに、データタイプを変えるとJSON形式の文字列で取得できることもわかった。
- watsonx.dataでは複数のデータソースのデータをSQL文で結合して取得できるので、既知の問題が直れば、watsonx.dataにMongoDBを組み込んで、MongoDBのデータと他のデータソースのデータとを結合して利用することができる。

#### 手順の流れ

1. MongoDBサービスを作成し、[MongoDB Compass](https://www.mongodb.com/ja-jp/products/tools/compass)で接続確認してみる
1. watsonx.dataに作成したMongoDBを組み込んでみる
1. オープンソースのPrestoを構築しMongoDBを組み込んでみる
1. PythonでPrestoにアクセスしMongoDBのデータを取得してみる

## 1. MongoDBサービスを作成し、MongoDB Compassで接続確認してみる
MongoDBサービスの作成は、IBM Cloudのカタログから`Databases for MongoDB`を作成した。

![mongodb_service.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/f8028fa6-27f5-5f0b-388e-660d8945a555.png)

しばらくすると利用可能となり、ステータスがアクティブとなる。次にMongoDBに接続するための認証情報を作成する。左側のメニューからService credentialsを選択し、右にあるNew credentialボタンをクリックして認証情報を作成する。

![mongodb_credentials.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/b23e7b4f-10b6-a37a-8f58-6ef5ddfd1668.png)

この情報のうち、以下の部分をMongoDB Compassからの接続に利用する。(.connection.mongodb.composed)
```
mongodb://ibm_cloud_36f4a076_6f69_4816_8036_13a190b7d741:a26f38ff9979d543b3fdff77212ae3caa8d3b69f18d6edaeaada1a1fb4ca7dfd@3d67c382-7065-45ad-84a9-19761d923525-0.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254,3d67c382-7065-45ad-84a9-19761d923525-1.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254,3d67c382-7065-45ad-84a9-19761d923525-2.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254/ibmclouddb?authSource=admin&replicaSet=replset&tls=true
```

書式は以下のようになっている。

```
mongodb://userid:password@
  host1:port,host2:port,host3:port/ibmclouddb?
  authSource=admin&replicaSet=replset&tls=true
```

あと、TLS/SSL認証に自己証明書を使っており、それも取得しておく。上記からも取得可能だが、左メニューの`Overview`を選択し、下の方にスクロールすると以下の画面が表示される。右の中程にある`Download Certificate`ボタンをクリックしてダウンロードする。名前は`mongodb1_tls.crt`とした。

![mongodb_certificate.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/661fb467-232a-9e59-da06-98a66bc4c29f.png)

自分のPCにMongoDB Compassを導入した。起動すると以下のような`New Connection`画面が表示されるので、`URI`に上記で認証情報から抜き出した情報を貼り付け、さらに、`Advanced Connection Options`を展開し、`TLS/SSL`タブにある`Select a file...`ボタンをクリックして、ダウンロードした`mongodb1_tls.crt`を登録する。

![compass_connection_settings.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/5a17706d-d600-51d1-082d-c7ff072f4dba.png)

あとは画面下にある`Save & Connect`ボタンをクリックすると、設定情報を保存し(保存する名前を聞かれたので`ibmcloud_mongodb`とした)、MongoDBに接続する。
接続すると以下の画面が表示される。`Databases`の右にある`+`をクリックし、今回のテストのために、データベース名`test`、コレクション名`shopinfo`を作成した。

![compass_connect_initial_page.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/f262e5bd-d3dc-2494-aea8-91a9481e5f7b.png)

以下のJSONデータをテスト用に作成したので、これをインポートする。
```json
$ cat shop.json 
{
    "id": "1",
    "name": "カフェA",
    "info": {
        "open": "8:00-21:00",
        "mail": {
            "info": "cafe-info@a.cafe.com",
            "reservation": "reservation@a.cafe.com"
        }
    },
    "review": [
        { "date": "2023-11-01", "comment": "おいしかった" },
        { "date": "2023-11-05", "comment": "落ち着いた雰囲気で良かった" }
    ]
}
{
    "id": "2",
    "name": "カフェB",
    "info": {
        "open": "7:00-20:00",
        "mail": {
            "info": "cafe-info@b.cafe.com",
            "reservation": "reservation@b.cafe.com"
        }
    },
    "review": [
        { "date": "2023-11-02", "comment": "メニューの種類が豊富で良かった" },
        { "date": "2023-11-03", "comment": "値段もリーズナブルだった" }
    ]
}
```

画面下部にある`Import Data`ボタンをクリックし、上記のファイルを指定する。

![mongodb_empty_collection.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/168e7c05-f1d8-3bf6-56d0-8f9cc6fcc2bf.png)

結果として以下のように2つのレコードがインポートされる。

![compass_imported_data.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/0416ae17-3e67-0aec-11e9-7819bd44172e.png)

## 2. watsonx.dataに作成したMongoDBを組み込んでみる
watsonx.dataにMongoDBを組み込むために、watsonx.dataの`インフラストラクチャー・マネージャー`画面の右側にある`コンポーネントの追加`を展開し、`データベースの追加`を選択する。

![watsonx_data_add_db.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/4f616385-3c3d-c7fc-830b-98aa0d1f5716.png)

以下のような`データベースの追加`画面が表示される。`データベース・タイプ`として`MongoDB`を選択し、各種入力項目をセットする。

![watsonx_data_add_db_panel.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/85610d67-83d5-9fe4-6380-fc100474d985.png)

その後、設定したMongoDBをPrestoエンジンに組み込むことで、PrestoエンジンからMongoDBにアクセスできるようになることになっているが、残念ながら既知の問題によりアクセスできない。問題はMongoDBの自己証明書の処理に関するもので、以下のエラーが発生する。
```
unable to find valid certification path to requested target
```

そこで、問題はいずれ解決されるはずなので、オープンソースのPrestoを構築してMongoDBを組み込み、MongoDBのJSONデータがPrestoのSQLでどのようにアクセスできるか確認してみた。

## 3. オープンソースのPrestoを構築しMongoDBを組み込んでみる
Prestoの構築は、以下のURLを参照した。(導入したバージョンは0.283)
- [Deploying Presto](https://prestodb.io/docs/current/installation/deployment.html)
- [MongoDB Connector](https://prestodb.io/docs/current/connector/mongodb.html)

MongoDBの自己証明書は、JKS(Java KeyStore)に組み込む必要がある。そのため、以下のコマンドでJKSファイルを作成するとともに、MongoDBの自己証明書を組み込む。
(MongoDBの自己証明書の別名を`mongodb1`とし、JKSのパスワードを`Passw0rd`とした)
```
keytool -import -trustcacerts -alias mongodb1 -file mongodb1_tls.crt -keystore presto.jks -storepass Passw0rd
```
この情報は、`etc/jvm.config`ファイルに以下のように指定する。(最後の2行)
```:etc/jvm.config
-server
-Xmx16G
-XX:+UnlockExperimentalVMOptions
-XX:+UseG1GC
-XX:G1HeapRegionSize=32M
-XX:+UseGCOverheadLimit
-XX:+ExplicitGCInvokesConcurrent
-XX:+HeapDumpOnOutOfMemoryError
-XX:+ExitOnOutOfMemoryError
-Djavax.net.ssl.trustStore=/xxx/presto-server-0.283/etc/presto.jks
-Djavax.net.ssl.trustStorePassword=Passw0rd
```

Prestoは複数のworkerノードで構成することができるが、ここではcoordinatorとworkerを1つのプロセスで動かす構成にした。構成ファイルは`etc/config.properties`ファイルで、ここにもJSKの情報を登録する。(最後の2行)
```:etc/config.properties
coordinator=true
node-scheduler.include-coordinator=true
http-server.http.port=8080
query.max-memory=50GB
query.max-memory-per-node=1GB
discovery-server.enabled=true
discovery.uri=http://localhost:8080
http-server.https.keystore.path=/xxx/presto-server-0.283/etc/presto.jks
http-server.https.keystore.key=Passw0rd
```

MongoDBの組み込みは`etc/catalog/mongodb.properties`ファイルに以下のように指定した。
```:etc/catalog/mongodb.properties
connector.name=mongodb
mongodb.seeds=3d67c382-7065-45ad-84a9-19761d923525-0.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254,3d67c382-7065-45ad-84a9-19761d923525-1.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254,3d67c382-7065-45ad-84a9-19761d923525-2.bqfh4fpt0vhjh7rs4ot0.databases.appdomain.cloud:32254
mongodb.credentials=ibm_cloud_36f4a076_6f69_4816_8036_13a190b7d741:a26f38ff9979d543b3fdff77212ae3caa8d3b69f18d6edaeaada1a1fb4ca7dfd@admin
mongodb.ssl.enabled=true
mongodb.socket-keep-alive=true
mongodb.socket-timeout=30000
mongodb.required-replica-set=replset
```
主なポイントは、mongodb.seedsに接続先のホスト名とポート番号のリスト、mongodb.credentialsにユーザーIDとパスワードの他に、authSource(認証情報を保持するコレクション名)としてadminを指定している。

また、Javaのバージョンは`OpenJDK build 21.0.1`だと以下のエラーが発生した。
```
Unable to load cache item
```
`java version "1.8.0_391"`に変えたところ正常に動作している。

## 4. PythonでPrestoにアクセスしMongoDBのデータを取得してみる
PythonでPrestoにアクセスするために、`presto-python-client`パッケージを導入し、`prestodb`をインポートして使用した。

presto-python-client`パッケージの導入
```
pip install presto-python-client
```

Prestoに組み込んだMongoDBにアクセスするプログラム
```python:presto_access_mongodb.py
import prestodb

if __name__ == '__main__':
    conn = prestodb.dbapi.connect(
        host='localhost',
        port=8080,
        user='admin',
        http_scheme='http',
        # auth=prestodb.auth.BasicAuthentication('admin','xxx') 
    )
    # conn._http_session.verify = '/certs/xxx.crt'
    cur = conn.cursor()
    cur.arraysize = 100 # 100がデフォルト
    cur.execute('select * from mongodb.test.shopinfo')
    while True:
        rows = cur.fetchmany()
        if len(rows) == 0:
            break
        for row in rows:
            print(row)
    cur.close()
    conn.close()
```

実行すると以下の結果が得られた。
```
$ python presto_access_mongodb.py
['1', 'カフェA', ['8:00-21:00', ['cafe-info@a.cafe.com', 'reservation@a.cafe.com']], [['2023-11-01', 'おいしかった'], ['2023-11-05', '落ち着いた雰囲気で良かった']]]
['2', 'カフェB', ['7:00-20:00', ['cafe-info@b.cafe.com', 'reservation@b.cafe.com']], [['2023-11-02', 'メニューの種類が豊富で良かった'], ['2023-11-03', '値段もリーズナブルだった']]]
```

各行で列の値が配列でセットされ、JSONの階層構造は配列を階層化してデータを保持していることがわかる。例えば、3番目の列は"info"キーの値を保持する。最初の行(レコード)の3番目のデータは、MongoDBでは以下のように保持されている。
```json
"info": {
    "open": "8:00-21:00",
    "mail": {
        "info": "cafe-info@a.cafe.com",
        "reservation": "reservation@a.cafe.com"
    }
}
```

パースして特定の情報を表示させるためには配列の要素数を指定すればよく、上記の3番目の列の値をパースするために、上記のプログラムの下記の部分を
```python
        for row in rows:
            print(row)
```

以下に置き換えてみる。
```python
        for row in rows:
            # print(row)
            print(f'オープン: {row[2][0]}')
            print(f'mail(情報): {row[2][1][0]}')
            print(f'mail(予約): {row[2][1][1]}')
```

すると以下の結果が得られる。
```
$ python presto_access_mongodb.py
オープン: 8:00-21:00
mail(情報): cafe-info@a.cafe.com
mail(予約): reservation@a.cafe.com
オープン: 7:00-20:00
mail(情報): cafe-info@b.cafe.com
mail(予約): reservation@b.cafe.com
```

MongoDBでJSON形式で保持されているのでJSONデータとして処理したいといった場合は、列のデータタイプを変更することでJSON形式の文字列が受け取れるので、そうすることによってPythonでJSONとして扱うことが可能となる。
Presto CLIでデータタイプを確認すると以下になっている。
```
$ ./presto --server localhost:8080
presto> describe mongodb.test.shopinfo;
 Column |                                  Type                                  | Extra | Comment 
--------+------------------------------------------------------------------------+-------+---------
 id     | varchar                                                                |       |         
 name   | varchar                                                                |       |         
 info   | row("open" varchar, "mail" row("info" varchar, "reservation" varchar)) |       |         
 review | array(row("date" varchar, "comment" varchar))                          |       |         
(4 rows)
```

MongoDBのデータタイプは、最初にPrestoからMongoDBの該当コレクションにアクセスするとPrestoが自動生成し、そのコレクションが属すDBの`_schema`コレクションにデータタイプを保存することがわかった。`_schema`コレクションには以下のような情報が保存されている。

![presto_mongodb_data_types.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/c079d225-7845-1ccd-e0a7-ebe3235b4b44.png)

この`_schema`コレクションに記載されている`info`と`review`の`type`を`varchar`にすると、JSON形式の文字列が受け取れるようになる。MongoDB Compassを使って変更するには、JSONデータ表示部分をクリックすると<img width="20" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/dae63ed1-1a7d-78d3-e7c0-cc47277de862.png">(ペンシル・アイコン)が表示されるので、それをクリックして変更し、画面下部に表示される`REPLACE`ボタンをクリックする。

![presto_mongodb_data_types_changed.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/507588/5e65949f-62b7-29a8-f568-567b7e3a2344.png)

Pythonのプログラムを最初の状態に戻して実行してみる。
```
$ python presto_access_mongodb.py
['1', 'カフェA', '{ "open" : "8:00-21:00", "mail" : { "info" : "cafe-info@a.cafe.com", "reservation" : "reservation@a.cafe.com" } }', '[{ "date" : "2023-11-01", "comment" : "おいしかった" }, { "date" : "2023-11-05", "comment" : "落ち着いた雰囲気で良かった" }]']
['2', 'カフェB', '{ "open" : "7:00-20:00", "mail" : { "info" : "cafe-info@b.cafe.com", "reservation" : "reservation@b.cafe.com" } }', '[{ "date" : "2023-11-02", "comment" : "メニュ\\u30fcの種類が豊富で良かった" }, { "date" : "2023-11-03", "comment" : "値段もリ\\u30fcズナブルだった" }]']
```
3番目の列と4番目の列がJSON形式に変わったことがわかる。上記の3番目の列の値をパースするために、上記のプログラムの下記の部分を
```python
        for row in rows:
            print(row)
            # print(f'オープン: {row[2][0]}')
            # print(f'mail(情報): {row[2][1][0]}')
            # print(f'mail(予約): {row[2][1][1]}')
```
以下に置き換えてみる。
```python
        for row in rows:
            # print(row)
            # print(f'オープン: {row[2][0]}')
            # print(f'mail(情報): {row[2][1][0]}')
            # print(f'mail(予約): {row[2][1][1]}')
            jinfo = json.loads(row[2])
            print(f'オープン: {jinfo["open"]}')
            print(f'mail(情報): {jinfo["mail"]["info"]}')
            print(f'mail(予約): {jinfo["mail"]["reservation"]}')        
```
さらに`import json`も追加する。
```python
import prestodb
import json

if __name__ == '__main__':
    ...
```

実行すると以下の結果が得られる。
```
$ python presto_access_mongodb.py
オープン: 8:00-21:00
mail(情報): cafe-info@a.cafe.com
mail(予約): reservation@a.cafe.com
オープン: 7:00-20:00
mail(情報): cafe-info@b.cafe.com
mail(予約): reservation@b.cafe.com
```

Pythonの場合は上記の2つの選択肢があるが、Javaの場合だとJSONデータを扱うのが大変なので階層化された配列で扱うのが良いと思う。