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

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

①シンプルな構造のJSON
②ネスト構造のJSON
③ネスト構造とリストを含むJSON
④ルートが読み出しルートではないJSON


■環境
Modeler 18.4
python 3.8.10
pandas 1.5.2
json 2.0.9
Windows 11

# サンプルストリーム
https://github.com/hkwd/210224PythonExtension/raw/master/230306json/json4.str

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/7e45b7c5-7ec6-8f54-779c-5d7172a8ec7c.png)

# サンプルデータ
https://github.com/hkwd/210224PythonExtension/tree/master/230306json

# 事前作業
以下の記事を参考にSPSSの拡張ノードでPythonを利用できるようにしておきます。

#### SPSS Modelerの拡張ノードでPythonを利用する（ミニマム設定） - Qiita 

https://qiita.com/kawada2017/items/d8f7db0cb1d15c1f47f4


## パッケージの導入
個人フォルダに作っている場合は普通にpip installできますが、All Userで利用できるPythonを"C:\Program Files\Python37"に導入しているので、管理者権限でpip installを行ってください。


```console:コマンド
python -m pip install pandas
python -m pip install json
```

# 拡張のインポートノードの作成

拡張のインポートノードをキャンバスに配置して、「シンタックス」タブで「Python for Spark」を選び、Pythonシンタックスに以下のコードをコピーペーストで貼り付けてください。
このスクリプトで4パターンのJSONファイルを読むことができます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/92f0feab-cf4e-d3eb-d4bc-d01788d03c13.png)

```python:JSONファイル読み込み.py
#入力JSONファイル名
jsonfile='C:/temp/json/01simple.json'
#df化したい辞書リストのルート要素
rowsroot = ""
#列名をフラット化する際の区切り文字
sep = '.'

#--------------------------------------
#JSON読込関数
import pandas as pd
import 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 flattenJsonFile(jsonfile, rowsroot, sep='.'):
    """
    JSONファイルを読み込み2次元のpandas DataFrameに変換する

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

    Returns
    -------
    df : pandas.DataFrame
        フラット化されたpandas DataFrame
    """
   # JSONファイルを読込
    with open(jsonfile, encoding='utf-8') as f:
        d = json.load(f)

    # df化したい辞書リストのルート項目を指定
    if rowsroot != '':
        d = d[rowsroot]

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

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

#Modelerで使用
import spss.pyspark.runtime
from pyspark.sql.types import *
asContext = spss.pyspark.runtime.getContext()
sqlContext = asContext.getSparkSQLContext()

# フラット化
df = flattenJsonFile(jsonfile, rowsroot, sep)

#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

#print(type(outputSchema))
#print(outputSchema)

asContext.setSparkOutputSchema(outputSchema)

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

# JSONファイルの読み込み
それぞれのパターンのファイルを読み込んでみます。


# ①シンプルな構造のJSONの読込

以下のようなシンプルなJSONはJSONファイルノードでも読むことができます。
```json:01simple.json
[
    {
        "id":1000,
        "UP_TIME":0,
        "POWER":948,
        "TEMP":250
    },
    {
        "id":1000,
        "UP_TIME":1,
        "POWER":945,
        "TEMP":251,
        "ERR_CD":1
    }
]
```

### JSONファイルノードで読み込む
読み込むことができます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/13169d0e-3ebb-0dca-4071-b28e0c485fc6.png)

### 拡張のインポートノードで読み込む

拡張のインポートノードでも読んでみます。
Pythonシンタックスで読み込むファイルを指定します。2行目のファイル名を「01simple.json」に設定してください。
```python:JSONファイル読み込み.py
#入力JSONファイル名
jsonfile='C:/temp/json/01simple.json'
```

JSONファイルノードと同様に読み込むことができますが、NULLの形式が異なります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/7c1d942d-7afa-e868-a6ff-39e4bf9ab729.png)

以下の置換ノードでSparkのNULLを「to_string(@FIELD)='nan'」で判定し、ModelerのNULL（undef）に置き換えます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/0ddfa62e-6e20-98b3-91ee-c6aaccae74a7.png)

同じように読み込めました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/f0340160-c8f6-5861-8de0-7b5c2cb89b09.png)


## ②ネスト構造を持つJSONの読込
次のデータはidの下にM_CDとUP_TIMEという2つのデータがぶら下がっています。

```json:02notnormal.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
        }
    }
]
```

### JSONファイルノードで読み込む
idの下にぶら下がっている、M_CDやUP_TIMEが一つの列の中に入ってしまいます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/d3494650-a562-9322-de36-aa1f1385f561.png)

### 拡張のインポートノードで読み込む

Pythonシンタックスで読み込むファイルを指定します。2行目のファイル名を「02notnormal.json」に設定してください。
```python:JSONファイル読み込み.py
#入力JSONファイル名
jsonfile='C:/temp/json/02notnormal.json'
```

やはり置換ノードでSparkのNULLを「to_string(@FIELD)='nan'」で判定し、ModelerのNULL（undef）に置き換えます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/9b0d888c-3b3b-9d58-b77e-42a379928d67.png)

「id」をid.M_CDとid.UP_TIMEの2列に切り出して読み込めました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/5dfc50a1-21a9-38af-9b12-723e68f7df91.png)


## ③ネスト構造とリストを含むJSON
次のデータは、ネスト化された辞書があるだけではなく、さらにcodeの下にERR_CDとMESSAGEのデータがリストで複数含まれています。
```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"
            }
        ]
    }
]

```


### JSONファイルノードで読み込む
codeの下のリストが一つの列の中に入ってしまいます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/61f75b8b-222e-0ef3-272f-97e3ad6e3a81.png)


### 拡張のインポートノードで読み込む

Pythonシンタックスで読み込むファイルを指定します。2行目のファイル名を「03list.json」に設定してください。
```python:JSONファイル読み込み.py
#入力JSONファイル名
jsonfile='C:/temp/json/03list.json'
```

やはり置換ノードでSparkのNULLを「to_string(@FIELD)='nan'」で判定し、ModelerのNULL（undef）に置き換えます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/c8515770-e3ee-c67d-030e-90220870f7ee.png)

code列が、'code.0.ERR_CD', 'code.0.MESSAGE', 'code.1.ERR_CD', 'code.1.MESSAGE'の4つの列としてフラット化されました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/1bc7705e-48d3-d28f-dc40-641249291375.png)


# ④ルートが読み出しルートではないJSONの読込
特に、REST APIでJSONデータを取得した場合の多くは、今まで見てきたようなリスト化されたJSONデータだけではなく、ヘッダーのような情報と組み合わせた一つの辞書型構造のデータであることが多いと思います。

例えば以下のようなJSONです。total_rowsという文書全体の属性をあらわすヘッダー的な項目があり、実際のデータはrowsの中のリストとして含まれています。

```json:04notroot.json
{"total_rows":3,"rows":
[
    {
        "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"
            }
        ]
    }
]
}
```

### JSONファイルノードで読み込む
rowsの下のJSONが一つの列の中に入ってしまいます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/392fe957-016c-98bc-a3e1-1e3c559932e7.png)


### 拡張のインポートノードで読み込む

Pythonシンタックスで読み込むファイルを指定します。2行目のファイル名を「04notroot.json」に設定してください。また、4行目の「rowsroot = "rows"」で"rows"
をルートとして読み込むという設定をします。

```python:JSONファイル読み込み.py
#入力JSONファイル名
jsonfile='C:/temp/json/04notroot.json'
#df化したい辞書リストのルート要素
rowsroot = "rows"
```

やはり置換ノードでSparkのNULLを「to_string(@FIELD)='nan'」で判定し、ModelerのNULL（undef）に置き換えます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/8521790d-23d6-a8c1-f215-5895dc542a60.png)


"rows"以下のJSONが行として展開されました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/2a007ae7-ffbf-c60c-0a0b-a4f7f6cbacb0.png)




# スクリプトの解説
以下で基本設定をしています。
jsonfileで入力ファイル名を指定しています。
rowsrootで読み出し始めるルートの要素を指定しています。
sepでネストした列を接続する際の区切り文字を指定しています。

```python:基本設定
#入力JSONファイル名
jsonfile='C:/temp/json/01simple.json'
#df化したい辞書リストのルート要素
rowsroot = ""
#列名をフラット化する際の区切り文字
sep = '.'
```

以下でflattenとflattenJsonFileという関数を定義しています。この解説は[このページ](https://qiita.com/kawada2017/items/937b6bd5ea81ea7974ad)を参考にしてください。


```python:JSON読込関数
#--------------------------------------
#JSON読込関数
---省略----
```

以下で定義したflattenJsonFile関数でJSONファイルをフラット化してPandasのDataFrameに変換しています。

```python:JSONファイルのフラット化
# フラット化
df = flattenJsonFile(jsonfile, rowsroot, sep)
```

PandasのBoolean型はSparkのDataFrameに変換できないのでstr型に変換しています。
```python:JSONファイルのフラット化
#Booleanはsparkdfに変換できないのでobjectはstrに変換
for col in df.columns:
    if df[col].dtype.name=='object':
        df[col]=df[col].astype(str)
```

PandasのDataFrameからSparkのDataFrameに変換しています。

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

asContext.setSparkOutputSchema(outputSchema)
````

SparkのDataFrameをModelerに返しています。
```python:SparkのDataFrameをModelerに返却
if not asContext.isComputeDataModelOnly():
    asContext.setSparkOutputData(outdf)
```


# 参考

#### SPSS ModelerのPython拡張ノード関連記事へのリンク - Qiita 
https://qiita.com/kawada2017/items/055ace72f9e1949004f3

#### JSONファイルをフラットなpandasデータフレームに変換 - Qiita 
https://qiita.com/kawada2017/items/937b6bd5ea81ea7974ad

#### Python for Spark を使用したスクリプト
https://www.ibm.com/support/knowledgecenter/ja/SS3RA7_18.2.2/modeler_r_nodes_ddita/clementine/r_pyspark_api.html