Modeler FlowでICOS上の複雑な構造のJSONファイルを読む

CP4DのModelerFlowでICOS上にある複雑な構造のJSONファイルを読んでみます。

ModelerFlowではICOS上のJSONファイルをJSONファイルノードで読み込むことができます。しかし、階層が深くなっていたり、繰り返し項目があったり、読み出しルートを変えたいということがあると、JSONファイルノードでは、読み込みたい形式で読めないことがあります。
そのため、ここではPythonの拡張ノードを使って、IOCS上のJSONファイルを読み込んでみます


* テスト環境
CP4D 4.0.9

* テストストリーム

https://github.com/hkwd/220201watsonstudio/raw/master/230330jsonicos/jsonicosmf.str

* テストデータ

 https://raw.githubusercontent.com/hkwd/220201watsonstudio/master/230330jsonicos/03list.json

# テストJSONファイル
ネスト化された辞書があるだけではなく、さらにcodeの下にERR_CDとMESSAGEのデータがリストで複数含まれている以下のデータを使います。このデータをICOSに置いています。

```json:03list.json
[
    {
        "id":{
            "M_CD":1000,
            "UP_TIME":0
        },
        "sensor":{"POWER":948,
            "TEMP":250
        }
    },
    {
        "id":{
            "M_CD":1000,
            "UP_TIME":1
        },
        "sensor":{"POWER":945,
            "TEMP":251
        },
        "code":[
            {
                "ERR_CD":1,
                "MESSAGE":"part1"
            }
        ]
    },
    {
        "id":{
            "M_CD":1000,
            "UP_TIME":2
        },
        "sensor":{"POWER":943,
            "TEMP":255
        },
        "code":[
            {
                "ERR_CD":2,
                "MESSAGE":"part2"
            },
            {
                "ERR_CD":3,
                "MESSAGE":"part3"
            }
        ]
    }
]

```

# 拡張ノードの実行


「拡張入力」ノードで、スクリプトタイプで「Spark用Python」を選び以下のようなPythonスクリプトを入力します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/9a628fb4-d8ed-db5e-db43-dfe07f557650.png)



```python:拡張入力
################################
# 初期設定

# ICOS設定
from pyspark.sql.types import *
import spss.pyspark.runtime
from ibm_botocore.client import Config
import ibm_boto3
import subprocess
import sys
import json
import pandas as pd
credentials = {
    "apikey": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "endpoints": "https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints",
    "iam_apikey_description": "Auto-generated for key crn:v1:bluemix:public:cloud-object-storage:global:a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:resource-key:2xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "iam_apikey_name": "200322icosbucket",
    "iam_role_crn": "crn:v1:bluemix:public:iam::::serviceRole:Writer",
    "iam_serviceid_crn": "crn:v1:bluemix:public:iam-identity::a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx::serviceid:ServiceId-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "resource_instance_id": "crn:v1:bluemix:public:cloud-object-storage:global:a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx::"
}

# Constants for IBM COS values
# 自分のエンドポイントに変更する
COS_ENDPOINT = "https://s3.jp-tok.cloud-object-storage.appdomain.cloud"

# ダウンロードするファイルが入っている Bucket名
source_bucket = "200322icosbucket"
# ダウンロードするファイル名(必要ならpathをつける)
source_file = "03list.json"

# --------------------------------------
# JSONフラット化設定
# df化したい辞書リストのルート要素
rowsroot = ""
# 列名をフラット化する際の区切り文字
sep = '.'


################################
# --------------------------------------
# JSON読込関数定義

# フラット化

def flatten(d, parent_key='', sep='.'):
    items = []
    for k, v in d.items():
        # 列名の生成
        new_key = parent_key + sep + k if parent_key else k
        # 辞書型項目のフラット化
        if isinstance(v, dict):
            items.extend(flatten(v, new_key, sep=sep).items())
        # リスト項目のフラット化
        elif isinstance(v, list):
            new_key_tmp = new_key
            for i, elm in enumerate(v):
                new_key = new_key_tmp + sep + str(i)
                # リストの中の辞書
                if isinstance(elm, dict):
                    items.extend(flatten(elm, new_key, sep=sep).items())
                # 単なるリスト
                else:
                    items.append((new_key, elm))
        # 値追加
        else:
            items.append((new_key, v))
    return dict(items)


def flattenJsonText(json_txt, rowsroot, sep='.'):
    """
    JSONファイルを読み込み2次元のpandas DataFrameに変換する

    Parameters
    ----------
    json_txt : string
        JSONのテキスト
    rowsroot : string
        フラット化するルートエレメント名。トップからでいい場合は空文字を入力する
    sep : string
        ノーマライズされていないエレメントを区切る文字

    Returns
    -------
    df : pandas.DataFrame
        フラット化されたpandas DataFrame
    """
    # JSONテキストを読込
    d = json.loads(json_txt)
    # df化したい辞書リストのルート項目を指定
    if rowsroot != '':
        d = d[rowsroot]

    # フラット化
    dlist = []
    for di in d:
        dlist.append(flatten(di, sep=sep))

    # フラット化された辞書をpandasデータフレームに変換
    return pd.DataFrame.from_dict(dlist)
# --------------------------------------
################################
# メイン処理


# ICOS接続
subprocess.check_call([sys.executable, '-m', 'pip', 'install', '--cache-dir',
                      '/tmp/.cache/pip', 'ibm-cos-sdk', '--quiet', '--no-input'])


cosclient = ibm_boto3.client("s3",
                             ibm_api_key_id=credentials['apikey'],
                             ibm_service_instance_id=credentials['resource_instance_id'],
                             # ibm_auth_endpoint=credentials['iam_url'],
                             config=Config(signature_version="oauth"),
                             endpoint_url=COS_ENDPOINT
                             )

# JSONデータの読み込み
json_txt = cosclient.get_object(Bucket=source_bucket, Key=source_file)[
    'Body'].read().decode('utf-8')

# フラット化
df = flattenJsonText(json_txt, rowsroot, sep)

# Modelerとのデータやり取り
asContext = spss.pyspark.runtime.getContext()
sqlContext = asContext.getSparkSQLContext()

# Booleanはsparkdfに変換できないのでobject型はstrに変換
for col in df.columns:
    if df[col].dtype.name == 'object':
        df[col] = df[col].astype(str)

# pandasdfからsparkdfに変換
outdf = sqlContext.createDataFrame(df)
outputSchema = outdf.schema


# スキーマ情報をModelerに設定
asContext.setSparkOutputSchema(outputSchema)

# SparkのDataFrameをModelerに返却
if not asContext.isComputeDataModelOnly():
    asContext.setSparkOutputData(outdf)

```

「テーブル」ノードを接続して実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/20ab8440-0d20-1dee-1c9f-f19e3c2127ca.png)


ICOS上のネストやリスト構造のあるデータの読み込みができました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/14b1654b-8931-4c03-ec54-021240eaea2b.png)

# コードの解説

以下のようにICOS接続のCredential、エンドポイント、bucket名、ファイル名とJSONルート要素、フラット化する際の区切り文字を指定します。
逆に言うとプロジェクト内の「接続」情報などは使えず、直接接続情報を記載する必要があります。
この部分にご自身の読みたいJSONファイルやそのルート要素などを指定します。

```python:初期設定
# ICOS設定
<中略>
credentials = {
    "apikey": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "endpoints": "https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints",
    "iam_apikey_description": "Auto-generated for key crn:v1:bluemix:public:cloud-object-storage:global:a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:resource-key:2xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "iam_apikey_name": "200322icosbucket",
    "iam_role_crn": "crn:v1:bluemix:public:iam::::serviceRole:Writer",
    "iam_serviceid_crn": "crn:v1:bluemix:public:iam-identity::a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx::serviceid:ServiceId-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    "resource_instance_id": "crn:v1:bluemix:public:cloud-object-storage:global:a/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx::"
}

# Constants for IBM COS values
# 自分のエンドポイントに変更する
COS_ENDPOINT = "https://s3.jp-tok.cloud-object-storage.appdomain.cloud"

# ダウンロードするファイルが入っている Bucket名
source_bucket = "200322icosbucket"
# ダウンロードするファイル名(必要ならpathをつける)
source_file = "03list.json"

# --------------------------------------
# JSONフラット化設定
# df化したい辞書リストのルート要素
rowsroot = ""
# 列名をフラット化する際の区切り文字
sep = '.'
```

これ以降はJSONファイルをただ読むというためであれば修正の必要はありません。

JSON読込関数定義を行っています。

```python:JSON読込関数
# --------------------------------------
# JSON読込関数定義

# フラット化

def flatten(d, parent_key='', sep='.'):
<中略>
def flattenJsonText(json_txt, rowsroot, sep='.'):
```

この関数の意味は以下で解説しています。

JSONファイルをフラットなpandasデータフレームに変換 - Qiita 

https://qiita.com/kawada2017/items/937b6bd5ea81ea7974ad#%E3%83%8D%E3%82%B9%E3%83%88%E6%A7%8B%E9%80%A0%E3%81%A8%E3%83%AA%E3%82%B9%E3%83%88%E3%82%92%E5%90%AB%E3%82%80json%E3%81%AE%E3%83%95%E3%83%A9%E3%83%83%E3%83%88%E5%8C%96

ICOSに接続するためにはpipでibm-cos-sdkをインストールする必要があるので以下で導入します。
```python:pip install ibm-cos-sdk
subprocess.check_call([sys.executable, '-m', 'pip', 'install', '--cache-dir',
                      '/tmp/.cache/pip', 'ibm-cos-sdk', '--quiet', '--no-input'])
```

ICOSに接続しJSONファイルを読み込みます。
```python:ICOSからJSON読込
cosclient = ibm_boto3.client("s3",
                             ibm_api_key_id=credentials['apikey'],
                             ibm_service_instance_id=credentials['resource_instance_id'],
                             # ibm_auth_endpoint=credentials['iam_url'],
                             config=Config(signature_version="oauth"),
                             endpoint_url=COS_ENDPOINT
                             )

# JSONデータの読み込み
json_txt = cosclient.get_object(Bucket=source_bucket, Key=source_file)[
    'Body'].read().decode('utf-8')
```

上で定義しておいたflattenJsonTextでJSONデータをpandas dataframeに変換します。
```python:JSONフラット化
# フラット化
df = flattenJsonText(json_txt, rowsroot, sep)
```

pandasのdataframeをSpark dataframeに変換してModelerの戻します。

```python:pandas dataframeをModelerにリターン
# Modelerとのデータやり取り
asContext = spss.pyspark.runtime.getContext()
sqlContext = asContext.getSparkSQLContext()

# Booleanはsparkdfに変換できないのでobject型はstrに変換
for col in df.columns:
    if df[col].dtype.name == 'object':
        df[col] = df[col].astype(str)

# pandasdfからsparkdfに変換
outdf = sqlContext.createDataFrame(df)
outputSchema = outdf.schema


# スキーマ情報をModelerに設定
asContext.setSparkOutputSchema(outputSchema)

# SparkのDataFrameをModelerに返却
if not asContext.isComputeDataModelOnly():
    asContext.setSparkOutputData(outdf)
```

# 参考情報
#### 参考：JSONファイルをフラットなpandasデータフレームに変換 - Qiita 

https://qiita.com/kawada2017/items/937b6bd5ea81ea7974ad

#### 参考：複雑な構造のJSONファイルをSPSS ModelerのPython拡張ノードで読み込む - Qiita 

https://qiita.com/kawada2017/items/801953b204db21f4e21e#%E3%82%B9%E3%82%AF%E3%83%AA%E3%83%97%E3%83%88%E3%81%AE%E8%A7%A3%E8%AA%AC

#### 参考：Python for Spark を使用したスクリプト - IBM Documentation 

https://www.ibm.com/docs/ja/cloud-paks/cp-data/4.0?topic=scripts-scripting-python-spark