PandasのSql Upsertを試してみた

## はじめに

多様なリソースからデータベースを構築するために，データ成形にはpandasを用いることが多いです．そのため
**pandasで作ったデータ -> DataBase**
というフローをスムーズに行うことが肝になります．

いままではreplaceとappendしかなかったため，pandas側で頑張って成形してから更新するしかありませんでした．(これがすごくめんどくさい)

そんな中，先日pandasに待望の[Sql Upsert機能追加のプルリク](https://github.com/pandas-dev/pandas/pull/29636)があったので，動かしてみました．これによって作業効率を爆上がりさせてくれるはずです・・！！

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/437488/c1713b7a-18c3-6cc0-9a54-4e1f79860c1e.png)

## そもそもUpsertとは

InsertとUpdateをやるという意味です．
SqlにおけるUpsertの機能は大きく2つあります．

1. Primary Keyベースで，存在するものはなにもせず，存在しないものはInsert(**upsert_keep**)
2. Primary Keyベースで，存在するものはUpdateして存在しないものはInsert(**upsert_overwrite**)

## なぜpandasでできるとうれしいか

* Sqlの欠損値の扱いが難しい
    * NULL文字列のみ欠損値と判断してくれる
        * PythonのnanやNone -> そのまま文字列としてテーブルに入ってしまう
        * 空白文字列 -> 0と認識される
* csv吐き出しがめんどくさい
    * せっかくPythonで書いてるのに，一旦csvに吐き出してから別のラッパーに突っ込むのは効率悪い

そのため，Pythonエンジニアにとっては**pandasから直接，安心してテーブル更新したい**という欲望があります．

## 結論

* Sql操作をほぼpandasのみでできるようになって便利．そしてシンプルで使いやすい
* upsert_keepが思うような挙動していないみたいなので原因調査する

```Python
'''
Attributes:
    df (pd.DataFrame): any DataFrame
    tablename (str): テーブル名
'''
import pandas as pd
from sqlalchemy import create_engine

con = create_engine('mysql+mysqlconnector://[user]:[pass]@[host]:[port]/[schema]')

# upsert_keep -> 基本なにもしない(?) バグの可能性もあるので今後みていく．
df.to_sql(tablename, con, if_exist='upsert_keep', index=False)

# upsert_overwrite -> 存在するものはUpdateして存在しないものはInsert．意図通り
df.to_sql(tablename, con, if_exist='upsert_overwrite', index=False)
```

## Env
ここからは具体的な話を書いていきます．
環境は以下です．

* macOS Mojave
    * 10.14.5
* Python -V
    * 3.6.8
* package version
    * SQLAlchemy 1.3.11

## Setting

環境構築

```shell
git clone https://github.com/V0RT3X4/pandas.git
cd pandas
git checkout -b sql-upsert
git pull origin  sql-upsert

git rev-parse HEAD # 現在のcommit hash値
# d0eb251075883902280cba6cd0dd9a1a1c4a69a4
```

### Installation from sources

```shell
pip install cython
# Successfully installed cython-0.29.14
python setup.py build_ext --inplace --force #結構時間かかります．気長に待ちましょう．
```

### Mysql server

localでのテスト用に，mysqlサーバを立てて，usersテーブルを作っておきます．

usersテーブル(**idカラムをprimary keyとする**)

| id | name | age |
|---:|---:|---:|
| 0 | taro | 20 |
| 1 | jiro | 17 |
| 2 | saburo | 18 |

```shell

mysql.server start --skip-grant-tables #　パスワード無しでログイン可能なmysqlサーバの起動
mysql #ログイン

mysql> CREATE DATABASE testdb;
mysql> USE testdb;
mysql> CREATE TABLE users (
         id INT NOT NULL,
         name VARCHAR(256),
         age INT
       );
mysql> ALTER TABLE users ADD PRIMARY KEY (id);
mysql> INSERT INTO users (id, name, age) VALUES
       (0, 'taro', 20),
       (1, 'jiro', 19),
       (2, 'saburo', 18);
```

## 本題

ここからはpandasとsqlalchemyを用いてSql Upsertを試していきます

### Db connect
まずはDBとの繋ぎこみから．sqlalchemyが必要になるので入っていない方は`pip install sqlalchemy`してください

```Python
from sqlalchemy import create_engine
import pandas as pd

con = create_engine('mysql+mysqlconnector://@localhost/testdb')
# format: 'mysql+mysqlconnector://[user]:[pass]@[host]:[port]/[schema]'
users = pd.read_sql('SELECT * FROM users;', con)
users
#   id    name  age
#0   0    taro   20
#1   1    jiro   19
#2   2  saburo   18
```

### Upsert

まずDataFrameを書き換えておく

```Python
users.loc[users.id==0, 'name'] = 'syo'
users = pd.concat([users, pd.DataFrame({'id': [3], 'name': ['shiro'], 'age': [28]})])
#   id    name  age
#0   0     syo   20
#1   1    jiro   19
#2   2  saburo   18
#0   3   shiro   28
```

to_sqlメソッドの基本構文は，

```Python
df.to_sql(tablename, con, if_exist, index=False)
# df: pd.DataFrame
# tablename: テーブル名

# if_exist: {'fail', 'replace', 'append', 'upsert_overwrite', 'upsert_keep'}

# * fail: Raise a ValueError. 
# * replace: Drop the table before inserting new values. 
# * append: Insert new values to the existing table. 
# * upsert_overwrite: Overwrite matches in database with incoming data. 
# * upsert_keep: Keep matches in database instead of incoming data.
```

(その他引数は[pandas/core/generic.py](https://github.com/V0RT3X4/pandas/blob/d0eb251075883902280cba6cd0dd9a1a1c4a69a4/pandas/core/generic.py#L2575)を確認！)

#### upsert_keep

* なにもしない(?) -> 後述
    * `3   shiro   28`をInsertしてくれると思ったけどされていない

```Python
users.to_sql('users', con, if_exists='upsert_keep', index=False)
pd.read_sql('SELECT * FROM users;', con) #確認
#   id    name  age
#0   0     taro   20
#1   1    jiro   19
#2   2  saburo   18
```

#### upsert_overwrite

* primary key(ここではid)かぶっていなければInsert, かぶっているレコードは更新する
    * 意図通り

```Python
users.to_sql('users', con, if_exists='upsert_overwrite', index=False)
pd.read_sql('SELECT * FROM users;', con) #確認
#   id    name  age
#0   0     syo   20
#1   1    jiro   19
#2   2  saburo   18
#3   3   shiro   28
```

## おわりに
### upsert_keepが意図する挙動にならない問題

[pandas/io/sql.py](https://github.com/V0RT3X4/pandas/blob/d0eb251075883902280cba6cd0dd9a1a1c4a69a4/pandas/io/sql.py#L695)の_upsert_keep_processingメソッドあたりをちゃんとみていく必要がありそう．原因わかり次第,プルリクand記事更新していきます．