Cloud Pak for DataのNotebookでDb2のデータを読み書きする

(2020/8/6) 7.複数レコード挿入でハマった問題と対応を追記

Cloud Pak for Data(以下CP4D)上で、分析プロジェクトのNotebookでPythonを使ってDb2のデータを読み書きする方法です。と言っても、CP4D固有の部分は最初のDB接続情報を取得するproject_libの部分だけで、後は一般的なpythonでのDb2の読み書きと同じです。

検証したCP4Dバージョン: v3.0.1

# CP4D上で使えるDb2接続用pythonライブラリ

代表的には3つの方法があるようです。いろいろ比較してみたところ、Read(Select)ならデータをPandasデータフレーム化してくれるpd.read_sqlが便利で、Write(Insert/Update)やその他DB操作はibm_dbを使うのが良さそうです。

| |pandas | ibm_db | ibmdbpy |
|:---|:---:|:---:|:---:|
|概要|select結果をpandasデータフレーム化(read_sql)<br>pandasデータフレームのデータを書き込み(to_sql)|SQLベタ書きなのでほぼ何でもできる<br>CP4Dに導入済|PythonからDb2を操作するライブラリの一種。CP4Dに導入済<br>[参考リンク](https://qiita.com/tetsuya1969/items/867ae78132e969fd078a)|
|レコード取得|○(read_sql)|○ (SELECT)<br>(*1)|○ (SELECT)<br>|
|テーブル洗い替え(全件消去&挿入)|○ (to_sql)<br>(*2)|○<br>(TRUNCATE→INSERT)|○<br>(TRUNCATE→INSERT)|
|レコード挿入|○ (to_sql)|○ (INSERT)<br>(*3)|○ (INSERT)|
|レコード更新&挿入|×|○ (MERGE)|× (*4)|

注記：

* (*1) 取得したデータをpandasデータフレーム化するにはフェッチする一工夫が必要。([参考リンク](https://www.ibm.com/support/knowledgecenter/ja/SSEPGG_9.7.0/com.ibm.swg.im.dbclient.python.doc/doc/t0054388.html))
* (*2) to_sqlの```if_exists = 'replace'```オプションで可能だが、テーブル名やスキーマ名を小文字で指定する必要があるなどクセがある。
* (*3) 複数レコードを挿入するときは、pandasデータフレームのデータをtuple化し、パラメーターマーカー化したSQL文を使う方法が便利。本文内でも紹介。([参考リンク](https://stackoverflow.com/questions/44957641/insert-into-db2-table-from-a-python-dataframe))
* (*4) 2020/7/6時点で、mergeに相当するオプションが無さそうです。

# Db2データを読み書きしてみる

## 1.DB接続情報を取得
ここがCP4D固有です。project_libを使って、予めデータ資産に登録したDB接続データ(テーブル毎に作成するもの)から接続情報を取得しています。TBL1がデータ資産名です。

```python
project = Project.access()
DB_credentials = project.get_connected_data(name="TBL1")
```

## 2. DB接続 (jaydebeapi)

pandasのread_sqlで使うDB接続を作成します。ここで1の接続情報を使います。パスワードなどをベタ書きする必要がないので安心です。

```python
import jaydebeapi
DB_connection_J = jaydebeapi.connect(
    'com.ibm.db2.jcc.DB2Driver',
    '{}://{}:{}/{}:user={};password={};'.format('jdbc:db2',
                                                DB_credentials['host'],
                                                DB_credentials['port'],
                                                DB_credentials['database'],
                                                DB_credentials['username'],
                                                DB_credentials['password'])
)
```

## 3. レコード取得 (pd.read_sql)
read_sqlでレコード取得すると、直ちにpandasデータフレーム化されるので便利です。

```python
import  pandas as pd
sql = "SELECT * FROM TT_SCHEMA.TBL1"
df_tbl1 = pd.read_sql(sql, DB_connection_J)
df_tbl1.head()
```

結果：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/69bcd9bd-3c4f-8830-c751-dbfe50060181.png)
TBL1は4列のDb2テーブルで、3行のレコードをサンプルとして含めています。
主キーはIDで、DEL_FLGはよくある論理削除用フラグをイメージしています。参考としてDDLを添付します。

```sql
CREATE TABLE TT_SCHEMA.TBL1 (
  ID INT NOT NULL,
  VALUE_INT INT,
  VALUE_CHAR VARCHAR(16),
  DEL_FLG INT NOT NULL default 0
)

ALTER TABLE TT_SCHEMA.TBL1 PRIMARY KEY (ID)
```

## 4. DB接続 (ibm_db)

ibm_dbでDB接続を作成します。ここでも1のDB接続情報を流用しています。

```python
import ibm_db
ssldsn = "DATABASE=" + DB_credentials['database'] + \
";HOSTNAME=" + DB_credentials['host'] + \
";PORT=" + str(DB_credentials['port']) + \
";PROTOCOL=TCPIP;UID=" + DB_credentials['username'] + ";PWD=" + DB_credentials['password'] + ";"

DB_connection_I = ibm_db.connect(ssldsn,'','')
```

## 5. レコード取得 (ibm_db)

ibm_dbでSELECT文でレコードを取得します。取得したデータをpandasデータフレームに入れるには、行ごとにfetchする必要がありちょっとめんどくさいです。pd.read_sqlの方がシンプル。 ([参考情報](https://www.ibm.com/support/knowledgecenter/ja/SSEPGG_9.7.0/com.ibm.swg.im.dbclient.python.doc/doc/t0054388.html))

```python
sql = "SELECT * FROM TT_SCHEMA.TBL1;"
result = ibm_db.exec_immediate(DB_connection_I, sql)

result_dict = ibm_db.fetch_assoc(result)
fields = list(result_dict.keys())
values = []
while result_dict != False:
    values.append(list(result_dict.values()))
    result_dict = ibm_db.fetch_assoc(result)

display(fields)
display(values)

df = pd.DataFrame(data = values, columns = fields)
display(df)
```

結果：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/412e4913-deac-6100-24f0-80fdff896229.png)

## 6. 1レコード挿入 (ibm_db)

ibm_dbでTBL1に1件挿入してみます。挿入前後のテーブルをpd.read_sqlで確認しています。

```python:before
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/60fc7514-42b1-24db-e291-187479979985.png)

```python:1件挿入
sql = "INSERT INTO TT_SCHEMA.TBL1 (ID, VALUE_INT, VALUE_CHAR, DEL_FLG) VALUES (4, 40, 'ddd', 0)"
result = ibm_db.exec_immediate(DB_connection_I, sql)
```

```python:after
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/5dc9ad7e-f035-b4b1-299f-e85a6202456d.png)

## 7. 複数レコード挿入 (ibm_db)

複数レコードを挿入する場合は、[こちらのStackoverflow記事](https://stackoverflow.com/questions/44957641/insert-into-db2-table-from-a-python-dataframe)にあるように、pandasデータフレームのデータをtuple化し、パラメーターマーカー化したSQL文を使う方法が便利です。

```python:before
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/60fc7514-42b1-24db-e291-187479979985.png)

```python:複数件挿入
# data to be inserted
df_insert = pd.DataFrame(data={'ID': [5,6,7], 'VALUE_INT': [50,60,70], 'VALUE_CHAR': ['eee','fff','ggg']})

# transform to tuple of tuples
tuple_of_tuples = tuple([tuple(x) for x in df_insert.values])

# insert multiple records
sql = "INSERT INTO TT_SCHEMA.TBL1 (ID,VALUE_INT,VALUE_CHAR) VALUES(?,?,?)"
stmt = ibm_db.prepare(DB_connection_I, sql)
result = ibm_db.execute_many(stmt, tuple_of_tuples)
```

```python:after
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/7251a148-abc5-b441-c9ba-3de4653f5e63.png)

(2020/8/6追記)
ibm_db.execute_manyで実行時に、```Value parameters array XX is not homogeneous with previous parameters array```というエラーにハマりました。
[こちらのStackoverflow記事](https://stackoverflow.com/questions/57964575/python-pandas-insert-dataframe-into-db2-table-using-ibm-db)には、NaNを置換すれば行けたよと書いてありましたが、それもNG。
結局、投入するpandas dataframeを作る前の元の値の一部が文字列(object)ではなく数値(int)だったことが原因でした。df.dtypesでよく確かめれば見つかる、データの型に起因する問題でした。

## 8. レコード更新＆挿入 (ibm_db)

TBL1の主キーはIDです。ID列が同じであればUpdate、なければInsertするという例です。
複数レコードの例を示しますが、1レコードでも使えます。

```python:before
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/60fc7514-42b1-24db-e291-187479979985.png)

ID=2をUpdate、5,6をInsertします。

```python:複数件Update&Insert
# data to be updated and inserted
df_insert = pd.DataFrame({'ID': [2,5,6], 'VALUE_INT': [200,50,60],'VALUE_CHAR':['aaa','bbb','ccc'],'DEL_FLG':[0,0,0]})

# transform to tuple of tuples
tuple_of_tuples = tuple([tuple(x) for x in df_insert.values])

# update & insert
sql="MERGE INTO TT_SCHEMA.TBL1 AS T1 USING (VALUES (?,?,?,?)) AS T2(ID,VALUE_INT,VALUE_CHAR,DEL_FLG) \
ON T1.ID=T2.ID \
WHEN MATCHED THEN UPDATE SET T1.VALUE_INT=T2.VALUE_INT,T1.VALUE_CHAR=T2.VALUE_CHAR, T1.DEL_FLG=T2.DEL_FLG \
WHEN NOT MATCHED THEN INSERT (ID,VALUE_INT,VALUE_CHAR,DEL_FLG) VALUES (T2.ID,T2.VALUE_INT,T2.VALUE_CHAR, T2.DEL_FLG);"  
stmt=ibm_db.prepare(DB_connection_I, sql)
result = ibm_db.execute_many(stmt,tuple_of_tuples)
```

```python:after
sql = "SELECT * FROM TT_SCHEMA.TBL1"
pd.read_sql(sql, DB_connection_J)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/397277/58b645f0-4cf4-64e3-505b-e103a619ce67.png)

## 最後にDisconnect
忘れずにDB接続を切断します。

```python
# disconnect jadebeapi connection
DB_connection_J.close()

# disconnect ibm_db connection
ibm_db.close(DB_connection_I)
```

以上です。
今後、気が向いたら他のパターンを試して追記します。