CP4D4.6のJupyternotebookでDBに書き出し、読み出し

CP4DではFlight serviceをつかうことで、APIKEYやパスワードをプログラム内に記述せずにDBなどのデータソースへの読み書きが可能です。読み出しのpythonコードは「コード」に挿入で簡単に生成ができますので、主に書き出しについて紹介します。


* テスト環境
    * CP4D v4.6.4
    * Python 3.10

* サンプルnotebook

https://github.com/hkwd/220201watsonstudio/blob/master/230609WriteTable/datawrite.ipynb

# 0. テストデータ

以下のようなテストデータをpandasのdataframeで用意しました。
```python
data = {
    'id': [1,2,3],
    'first name': ['Suzuki', 'Sato', 'Takahashi']
}

df = pd.DataFrame(data)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/530c9984-fe6f-6357-b9c2-12a05e28d863.png)


# 1.DB書込
ここではDB2に書き込んでみます。

あらかじめ、接続の資産登録をしておきます。この例では「DBWHOC＿HDM」を使います。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223763/bd25f8de-dcbd-b173-653a-33f84d625c7a.png)

以下のコードで書き込みが可能です。

```python
import itc_utils.flight_service as itcfs
import pyarrow as pa

# create an instance of a flight client
flightClient = itcfs.get_flight_client()

flight_data_request = {
    'connection_name': """DB2WHOC_HDM""",   # name of the connection
    #'commit_frequency':  5000 optional parameter to trigger a commit every n records
    'interaction_properties': {
        'schema_name': 'HKWD', 
        'table_name': 'FLIGHTTEST',
        'table_action': 'replace' # truncate|append
        #'create_statement' : 'CREATE TABLE ...' # if you want to control data types
    },
}



# create a flight descriptor specifiying the data source or target
# content and structure of cmd are specific to IBM CP4D's flight service
flightDescriptor = itcfs.get_flight_descriptor(nb_data_request = flight_data_request)

# obtain a flight write stream
schema = pa.Schema.from_pandas(df, preserve_index=False)
writer, reader = flightClient.do_put(flightDescriptor, schema)

# write data to a data target
writer.write_table(pa.Table.from_pandas(df, schema))
writer.close()
```

解説をします。
Flight serviceとpyarrowのパッケージを読み込み、Flight serviceのクライアントインスタンスを作ります。
```python
import itc_utils.flight_service as itcfs
import pyarrow as pa
# create an instance of a flight client
flightClient = itcfs.get_flight_client()
```

以下が肝の部分です。
connection_nameには資産登録した接続名を設定します。この例では「DBWHOC＿HDM」でした。
次に、interaction_propertiesの中にスキーマ名、テーブル名を設定し、そしてtable_actionでここでは「replace」を指定しました。行追加の場合は「append」になります。
そしてこの辞書型データからflightDescriptorオブジェクトを作っています。

```python
flight_data_request = {
    'connection_name': """DB2WHOC_HDM""",   # name of the connection
    #'commit_frequency':  5000 optional parameter to trigger a commit every n records
    'interaction_properties': {
        'schema_name': 'HKWD', 
        'table_name': 'FLIGHTTEST',
        'table_action': 'replace' # truncate|append
        #'create_statement' : 'CREATE TABLE ...' # if you want to control data types
    },
}

# create a flight descriptor specifiying the data source or target
# content and structure of cmd are specific to IBM CP4D's flight service
flightDescriptor = itcfs.get_flight_descriptor(nb_data_request = flight_data_request)

```

interaction_propertiesのオプションは[ここ](https://api.dataplatform.cloud.ibm.com/v2/data_flows/doc/dataasset_and_connection_properties.html#dashdb)に載っています。RDBの種類やデータソースの種類によっても異なるプロパティが用意されていますが、ここではDb2の書き込み時のオプションを自動翻訳しています。

CREATE TABLE文を指定したり、MERGEなどのSQLを作成することもできるようです。

| Name               |                                                                                    Description                                                                                   |
|--------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------:|
| create_statement   | ターゲットテーブルを再作成するための   Create DDL ステートメント                                                                                                                 |
| key_column_names   | 更新またはマージ中に使用される主キーをオーバーライドするための列名のカンマ区切りリスト。                                                                                         |
| query_timeout      | クエリのタイムアウトを指定します。   指定しない場合は、デフォルト値の 300 秒または 5 分が使用されます。                                                                          |
| schema_name        | 書き込むテーブルが含まれるスキーマの名前                                                                                                                                         |
| static_statement * | セットアップ操作に使用される   SQL (CREATE ステートメントなど)                                                                                                                   |
| table_action       | 新しいデータセットを処理するためにターゲットテーブルに対して実行するアクション。   値: [append, replace, truncate]。 デフォルト: append                                                 |
| table_name         | 書き込むテーブルの名前                                                                                                                                                           |
| update_statement   | テーブル内のデータを更新するための   SQL INSERT、UPDATE、MERGE、または DELETE ステートメント。                                                                                   |
| write_mode         | ターゲットテーブルにレコードを書き込むためのモード。   値: [insert, merge, static_statement, update, update_statement, update_statement_table_action]。 デフォルト: insert |

元データのdataframeからスキーマオブジェクトをつくり、flightDescriptorとあわせて、writerオブジェクトを作っています（readerも一緒につくられますが、使いません。）。

```python
# obtain a flight write stream
schema = pa.Schema.from_pandas(df, preserve_index=False)
writer, reader = flightClient.do_put(flightDescriptor, schema)
```
write_tableメソッドでdataframeとschemaを指定して書き出します。
```python
# write data to a data target
writer.write_table(pa.Table.from_pandas(df, schema))
writer.close()
```

# 2.DB読込

書き込んだデータを読み出して確認します。
やはりflight_data_requestを作成します。
```python
flight_data_request = {
    'connection_name': """DB2WHOC_HDM""",
    'interaction_properties': {
        'schema_name': 'HKWD', 
        'table_name': 'FLIGHTTEST'
        #'select_statement': 'SELECT * FROM "HKWD"."FLIGHTTEST" FETCH FIRST 5000 ROWS ONLY'
    }
}

flightInfo = itcfs.get_flight_info(flightClient, nb_data_request=flight_data_request)

data_df_1 = itcfs.read_pandas_and_concat(flightClient, flightInfo, timeout=240)
data_df_1
```

これも先ほどの[リンク](https://api.dataplatform.cloud.ibm.com/v2/data_flows/doc/dataasset_and_connection_properties.html#dashdb)にinteraction_propertiesで指定できるプロパティが載っています。
| Name                  |                                                                     Description                                                                    |
|-----------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------:|
| byte_limit            | 返される最大バイト数。   単位を変更するには、KB、MB、GB、または TB のいずれかの接尾辞を使用します。 値 0 はすべてのデータを返します。              |
| call_statement        | ストアド   プロシージャを実行する SQL ステートメント                                                                                               |
| decimal_rounding_mode | 10   進数および数値データ型の列の値に対するさまざまな丸めモード。 値: [ceiling, down, floor, halfdown, halfeven,   halfup, up]。 デフォルト: floor |
| query_timeout         | クエリのタイムアウトを指定します。   指定しない場合は、デフォルト値の 300 秒または 5 分が使用されます。                                            |
| read_mode             | テーブルからレコードを読み取るメソッド。   値: [general, select]。 デフォルト: general                                                             |
| row_limit             | 返される行の最大数                                                                                                                                 |
| schema_name           | 読み取るテーブルが含まれるスキーマの名前                                                                                                           |
| select_statement *    | テーブルからデータを取得するための   SQL SELECT ステートメント                                                                                     |
| table_name *          | 読み取るテーブルの名前                                                                                                                             |


# 参考
#### Python ノートブックの Flight service - IBM Documentation 

https://www.ibm.com/docs/ja/cloud-paks/cp-data/4.6.x?topic=notebook-flight-service-in-python-notebooks


#### IBM Watson Data Platform Data Flows Service - Data Asset and Connection Properties 

https://api.dataplatform.cloud.ibm.com/v2/data_flows/doc/dataasset_and_connection_properties.html