CP4DのWatson StudioのJupyterからSQLAlchemyでDb2を操作する

CP4DのWatson StudioのJupyterからDb2を操作します。
CP4Dの接続オブジェクトからDb2への接続情報を取得して、SQLAlchemyアダプターのibm_db_saで接続します。

• テスト環境
  • CP4D 4.0.5
  • db2warehouse on cloud
  • python　3.8.11
  • pandas 1.2.4
  • ibm-watson-studio-lib 3.0.6
  • SQLAlchemy 1.4.22
  • ibm_db_sa 0.3.7

1 接続オブジェクトのリストを取得

Watson StudioのプロジェクトにDb2への接続オブジェクト（例：DB2WHOC_HDM）を登録しておきます。

接続オブジェクトにはDb2に接続するための情報が記録されています。

ibm_watson_studio_libでPythonコードから接続オブジェクトにアクセスが可能です。

list_connectionsで接続オブジェクトのリストを得ることができます。

接続オブジェクトのリスト

import ibm_watson_studio_lib
wslib = ibm_watson_studio_lib.access_project_or_space()
# 接続オブジェクトのリスト
assets=wslib.list_connections()
wslib.show(assets)

get_connectionで接続オブジェクトの設定を取得できます。database, hostname, username, passwordなどの情報が取れます。

接続オブジェクトの情報を取得

connection_config=wslib.get_connection('DB2WHOC_HDM')
connection_config

2 Db2への接続

Db2を扱うドライバーやアダプターはいろいろありますが、ここではpandasへの読み書きが容易なSQL Alchemyのアダプターであるibm_db_saをつかってDb2へ接続します。

• 参考:Python ドライバー、SQLAlchemy アダプター、Django アダプターのインストールの検証 - IBM Documentation

SQLAlchemyで接続

import sqlalchemy
from sqlalchemy import *
import ibm_db_sa
con_params = 'ibm_db_sa://%s:%s@%s:%s/%s' % (
    connection_config['username'], connection_config['password'], connection_config['host'], connection_config['port'], connection_config['database'])
if connection_config['ssl']=='true' :
    con_params = con_params+ ";Security=SSL"
con_sa = sqlalchemy.create_engine(con_params)

3 Db2からSELECT

SELECT結果をpd.read_sqlでpandasのDataframeに読み込んでみます。

データフレームに検索結果の読み込み

import pandas as pd
# SELECT文の実行
sql = "select 2 as ID, 'suzuki' as NAME, 0.1 as RATE, SYSDATE as TDATE from SYSIBM.SYSDUMMY1" \
   " union all select 1 as ID, 'sato' as NAME, 2.5 as RATE, SYSDATE as TDATE from SYSIBM.SYSDUMMY1"
df = pd.read_sql(sql, con_sa)

データタイプを確認しておきます。文字列型はobject型になっています。

データタイプの確認

df.dtypes

4 Db2のテーブルへの書き込み

DataFrameをテーブルとして書き込みます。何度も実行することを想定して、すでに表がある場合はDROPしてCREATEしなおす、洗い替えをするコードを書きました。

if_exists='replace'が動作しないので削除をしておく

sql = text('DROP TABLE IF EXISTS hkwddummy;')
result = con_sa.execute(sql)

to_sqlは、DataFrameのデータ型に従って、CREATE TABLEとINSERTを自動で行ってくれます。ただ、object型のままだとDb2上でCLOBのデータ型になってしまったので、String(255) にマッピングしなおしました。
最適なサイズにするためには、きっちりとした型指定を行ってください。

dfのテーブルへの書き込み

＃object型をCLOB型にしないためにVARCHAR(255)で指定。
cols = df.dtypes[df.dtypes=='object'].index
type_mapping = {col : String(255) for col in cols }
df.to_sql(name='hkwddummy', con=con_sa, schema='hkwd', dtype=type_mapping)

• 参考：sqlalchemy/pandas - sqlalchemy読み込み用のコラムをCLOBとして使用する - jpndev

テーブルのスキーマ情報は以下で確認ができます。

作成されたテーブルのスキーマ情報の確認

inspector = sqlalchemy.inspect(con_sa)
inspector.get_columns("hkwddummy")

• 参考：SQLAlchemyを利用して既存テーブルのメタデータを取得する - Qiita

5 接続破棄

破棄

con_sa.dispose()

サンプルコード

参考

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

watson studioのJupyter notebookからdb2へアクセス