SQLエンジニアのためのPandas入門その2

はじめに

前回はPandasの使い方をSQL文との比較で書いてみましたが
sqliteを利用してSQL文そのままの記述でPandasのデータが扱えたので
まとめてみました

PythonでCSVを読み込む際に、pandasとSQLite3を使ってCSVをSQLで操作する方法の備忘録です。

前準備

• pandas　
  データ解析機能を提供するpythonライブラリ
  pipでpandasをインストール

command line

pip install pandas
# または　pip3 install pandas

• SQLite3
  SQLiteを操作する機能を提供するpythonライブラリ
  SQLite3はPythonの標準ライブラリに含まれています。

実行

CSVを読み込んでDB、テーブルを作成

import pandas 
import sqlite3
 
# CSVファイルをDataFrameとして読み込む
df = pandas.read_csv("./test.csv")
# メモリ上にSQLiteデータベースを作成(一時的に)
conn = sqlite3.connect(":memory:")
 
# テーブルのCREATE文を作成
table_name = "temp_table" 
df.to_sql(name=table_name, con=conn, if_exists="replace", index=False)

1. df = pandas.read_csv("./test.csv")
pandasを使ってCSVをDataFrameとして読み込んでいます。

主要なread_csvパラメータ一覧

パラメータ名	パラメータ詳細	型	デフォルト値
filepath_or_buffer	読み込むCSVファイルのパス	strまたはread()関数が使われているオブジェクト	(必須)
sep/delimiter	区切り文字の指定	str	","
index_col	インデックス列の指定	int,sequence,False	False
header	ヘッダー(列名)行の指定	int,intの配列,None, "infer"	"infer"
names	列(カラム)名の指定	配列,None	None
usecols	読み込むカラムの指定	strの配列もしくは呼び出し可能なもの	None
dtype	データもしくは列(カラム)ごとの型を指定	型名もしくはカラムの辞書 (例：{"列名1": "int64"})	None
skiprows	読み飛ばす行の指定	int,intの配列,None	None
nrows	読み込む行数の指定	int,None	None

("infer"は、names引数があればNone、names引数がなければ0と同じになります)

2. conn = sqlite3.connect(":memory:")
sqlite3でメモリ上に一時的にSQLiteデータベースを作成、接続します。
(処理が終わったら破棄されます)

3. df.to_sql(table_name, conn, if_exists="replace", index=False)
DataFrameとして読み込んだCSVを元にテーブルを作成します。

主要なto_sqlパラメータ一覧

パラメータ名	パラメータ詳細	型	デフォルト値
name	作成するテーブルの名称	str	(必須)
con	接続するデータベース	Connection	(必須)
if_exists	テーブルが既に存在する場合の動作	"fail", "replace", "append"	"fail"
index	DataFrameのインデックス列をテーブルに書き込むかどうか	bool	True
chunksize	一度に書き込まれる行数	int	None
dtype	列(カラム)ごとの型を指定	dict (例：{"列名1": "int64"})	None

作成したDBでSQL文を使用する

# SELECT文
# クエリを実行してDBにあるデータをDataFrameに格納
result = pandas.read_sql_query("SELECT * FROM temp_table", conn)
print(result) # 表示
 
# INSERT文,UPDATE文,DELETE文
query = "INSERT INTO temp_table (id, row1, ...) VALUES (3, 1, ...)"
query = "UPDATE temp_table SET row1 = 3 WHERE id = 3"
query = "DELETE FROM temp_table WHERE id = 3"
# SQL文の実行
cursor = conn.cursor()
cursor.execute(query)
conn.commit()
 
# 変更を反映したデータを再読み込み
new_df = pandas.read_sql_query("SELECT * FROM temp_table", conn)
# 新しいDataFrameをCSVファイルに書き込む
new_df.to_csv("./test.csv", index=False)

• SELECT文の場合
  pandas.read_sql_queryで作成したDBからデータを読み込む
• INSERT文,UPDATE文,DELETE文の場合
  cursor = conn.cursor() cursor.execute(query) conn.commit()で作成したDBにクエリを実行して、変更されたDBをCSVに書き込む

主要なread_sql_queryパラメータ一覧

パラメータ名	パラメータ詳細	型	デフォルト値
sql	実行するSQLクエリ文字列	str	(必須)
con	接続するデータベース	Connection	(必須)
index_col	インデックスとして使用する列名	str,strの配列,None	None
coerce_float	数値を浮動小数点数に強制的に変換するかどうか	bool	True
parse_dates	日付として解析する列のリスト	strの配列,None	None
chunksize	データを分割して読み込む際のチャンクサイズ	int	None
dtype	データもしくは列(カラム)ごとの型を指定	型名もしくはカラムの辞書 (例：{"列名1": "int64"})	None

主要なto_csvパラメータ一覧

パラメータ名	パラメータ詳細	型	デフォルト値
path_or_buf	出力先のファイルパスまたはファイルオブジェクト	str	(必須)
sep	区切り文字の指定	str	","
na_rep	欠損値の表現方法	str	""
float_format	浮動小数点数のフォーマット指定	str	True
columns	出力する列の指定	sequence	None
header	ヘッダーの出力有無または指定	bool,list,str	True
index	インデックスの出力有無	bool	True
index_label	インデックス列のラベル	str,sequence,False	None
chunksize	データを分割して読み込む際のチャンクサイズ	int	None

まとめ

実際に動かすとしたら、こんな感じになると思います。

import pandas 
import sqlite3
 
# 共通のテーブル名
TABLE_NAME = "temp_table"
 
 
# CSVをDB、テーブルに変換
def csv_to_db(path: str):
    # CSVファイルをDataFrameとして読み込む
    df = pandas.read_csv(path)
    # メモリ上にSQLiteデータベースを作成(一時的に)
    conn = sqlite3.connect(":memory:")
    # テーブルのCREATE文を作成
    df.to_sql(name=TABLE_NAME, con=conn, if_exists="replace", index=False)
    #接続情報を返す
    return conn
 
 
# SELECT文実行
def select_sql(path: str, query: str):
    with csv_to_db(path) as conn:
        result = pandas.read_sql_query(query, conn)
    # 結果を返す(DataFrame)
    return result
 
 
# INSERT文,UPDATE文,DELETE文実行
def execute_sql(path: str, query: str):
    with csv_to_db(path) as conn:
        cursor = conn.cursor()
        cursor.execute(query)
        conn.commit()
        # 変更を反映したデータを再読み込み
        new_df = pandas.read_sql_query(f"SELECT * FROM {TABLE_NAME}", conn)
        # 新しいDataFrameをCSVファイルに書き込む
        new_df.to_csv(path, index=False)
 
 
# INSERT文,UPDATE文,DELETE文を使う場合
query = f"INSERT INTO {TABLE_NAME} (id, row1, ...) VALUES (3, 1, ...)"
query = f"UPDATE {TABLE_NAME} SET row1 = 3 WHERE id = 3"
query = f"DELETE FROM {TABLE_NAME} WHERE id = 3"
execute_sql("./test.csv", query)
 
# SELECT文を使う場合
query = f"SELECT * FROM {TABLE_NAME}"
result_df = select_sql("./test.csv", query)
print(result_df)
 

以上、pandasとSQLite3を使ってCSVをSQLで操作する方法でした

参考にさせていただいた記事、サイト