Pandas で SQL データベースを読み取る

Pandas は SQL データベースと直接やり取りする一連の関数を提供しており、クエリ結果を直接 DataFrame として読み取ったり、DataFrame をデータベースに書き戻したりできます。これにより、データアナリストはデータベース接続や結果の解析を手動で処理する必要がなくなり、データベースと Python のやり取りの流れが大幅に簡素化されます。

主要関数の概要

関数 用途 戻り値の型
pd.read_sql() SQL クエリを実行するか、テーブル全体を読み取ります（両方のシナリオに対応した汎用関数） DataFrame
pd.read_sql_query() SQL クエリ文を実行し、複雑なクエリに適しています DataFrame
pd.read_sql_table() テーブル全体を直接読み取ります。SQLAlchemy 接続のみサポートします DataFrame
DataFrame.to_sql() DataFrame をデータベーステーブルに書き込みます None / int

実務では推奨しますpd.read_sql()、渡された引数に応じてクエリの実行かテーブル全体の読み取りかを自動的に判断するため、互換性が最も優れています。pd.read_sql_table()SQLAlchemy エンジン接続のみをサポートし、ネイティブのsqlite3などの DB-API 接続はサポートしません。

データベース接続の確立

Pandas 自体はデータベースに直接接続せず、サードパーティライブラリを使用して接続を確立してから渡す必要があります。主流の方法は2つあります：SQLAlchemy エンジン（推奨）とDB-API ネイティブ接続（軽量でシンプル）。

方法1: SQLAlchemy（推奨）

SQLAlchemy は Python で最も主流のデータベースツールキットであり、すべての主要なデータベースをサポートし、Pandas との互換性も最も優れています。

pip install sqlalchemy

例

from sqlalchemy import create_engine

# SQLAlchemy 接続文字列の形式: データベースタイプ+ドライバー://ユーザー名:パスワード@ホスト:ポート/データベース名
# SQLite（ファイルデータベース、ユーザー名・パスワード不要）
engine = create_engine("sqlite:///mydata.db")

# MySQL
engine = create_engine("mysql+pymysql://root:password@localhost:3306/mydb")

# PostgreSQL
engine = create_engine("postgresql+psycopg2://user:password@localhost:5432/mydb")

# SQL Server
engine = create_engine("mssql+pyodbc://user:password@server/mydb?driver=ODBC+Driver+17+for+SQL+Server")

# 接続が成功したか検証する
with engine.connect() as conn:
    print("接続成功")

方法2: sqlite3 ネイティブ接続（SQLite のみ）

Python 組み込みsqlite3モジュールで、追加インストールは不要で、軽量なローカル SQLite データベースに適しています：

例

import sqlite3

# SQLite ファイルデータベースに接続（ファイルが存在しない場合は自動作成）
conn = sqlite3.connect("mydata.db")

# インメモリデータベースに接続（プログラム終了後にデータは消えるため、テストに最適）
conn_memory = sqlite3.connect(":memory:")

# 使用後は手動で接続を閉じる必要がある
# conn.close()

pd.read_sql() でデータを読み取る

基本構文

pd.read_sql(sql, con, index_col=None, coerce_float=True, params=None, parse_dates=None, columns=None, chunksize=None)

主なパラメータの説明:

パラメータ 型 説明
sql str SQL クエリ文、またはテーブル名（conタイプに関係します）
con 接続オブジェクト SQLAlchemy エンジンまたは DB-API 接続オブジェクト
index_col str または list 指定した列を DataFrame の行インデックスに設定します
params list または dict SQL パラメータ化クエリのパラメータ値で、SQL インジェクションを防ぎます
parse_dates list または dict 指定した列を datetime 型として解析します
chunksize int チャンク単位で読み取り、各チャンクは指定行数の DataFrame イテレータを返します

1、テーブル全体を読み取る

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# employees テーブル全体を読み取る
df = pd.read_sql("employees", con=engine)
print(df.head())
print(f"合計 {len(df)} 行、{len(df.columns)} 列")

2、SQL クエリを実行する

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# 条件付きクエリ
df = pd.read_sql("SELECT * FROM employees WHERE department = 'IT'", con=engine)

# 複数テーブルの結合クエリ
sql = """
    SELECT e.name, e.salary, d.department_name
    FROM employees e
    JOIN departments d ON e.dept_id = d.id
    WHERE e.salary > 10000
    ORDER BY e.salary DESC
"""
df = pd.read_sql(sql, con=engine)
print(df.head(10))

3、パラメータ化クエリ（SQL インジェクション防止）

クエリ条件がユーザー入力からの場合、必ずパラメータ化クエリを使用してください、文字列連結で SQL を作成しないでください：

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# ❌ 危険な書き方：文字列連結 SQL には SQL インジェクションのリスクがあります
dept = "IT"
# df = pd.read_sql(f"SELECT * FROM employees WHERE department = '{dept}'", engine)

# ✅ 安全な書き方：? プレースホルダ（sqlite3）または :name 名前付きパラメータ（SQLAlchemy）を使用
# SQLite / DB-API スタイル（? を使用）
df = pd.read_sql(
    "SELECT * FROM employees WHERE department = ? AND salary > ?",
    con=engine,
    params=["IT", 8000]   # params はリストで、位置に応じて ? を置き換えます
)

# SQLAlchemy 名前付きパラメータスタイル（:param_name を使用）
from sqlalchemy import text
with engine.connect() as conn:
    df = pd.read_sql(
        text("SELECT * FROM employees WHERE department = :dept AND salary > :min_salary"),
        con=conn,
        params={"dept": "IT", "min_salary": 8000}
    )

print(df)

4、インデックス列を設定する

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# データベースの id 列を DataFrame の行インデックスに設定する
df = pd.read_sql("SELECT * FROM employees", con=engine, index_col="id")
print(df.head())

# 複数の列を複合インデックスとして使用する
df = pd.read_sql(
    "SELECT * FROM orders",
    con=engine,
    index_col=["year", "month"]  # 複合インデックス
)

5、日付列を解析する

データベースに保存されている日付フィールドは、読み取り後はデフォルトで文字列型です。parse_datesパラメータを使用すると直接解析できます。datetime：

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# created_at と updated_at 列を datetime 型として解析する
df = pd.read_sql(
    "SELECT * FROM orders",
    con=engine,
    parse_dates=["created_at", "updated_at"]
)

print(df.dtypes)
# created_at    datetime64[ns]
# updated_at    datetime64[ns]

# 解析形式を指定することもできます（標準以外の日付形式向け）
df = pd.read_sql(
    "SELECT * FROM orders",
    con=engine,
    parse_dates={"created_at": "%Y%m%d"}  # "20240115" のような形式を日付として解析します
)

大量データのチャンク読み取り（chunksize）

データベーステーブルのデータ量が非常に多い場合、一度にすべてメモリに読み込むと OOM（メモリ不足）が発生する可能性があります。chunksizeパラメータを使用すると、データを分割して読み取り、毎回一部分だけを処理できます：

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:password@localhost/bigdata")

# chunksize=10000 は毎回 10000 行を読み取り、イテレータを返すことを意味します
chunks = pd.read_sql("SELECT * FROM large_table", con=engine, chunksize=10000)

# チャンクごとに処理（毎回 10000 行のみがメモリ内にあります）
result_list = []
for i, chunk in enumerate(chunks):
    # 各チャンクのデータに対して処理ロジックを実行（例: フィルタリング、集計など）
    processed = chunk[chunk["status"] == "active"]
    result_list.append(processed)
    print(f第 {i+1} チャンクを処理しました。現在のチャンクの有効行数: {len(processed)})

# 処理済みのすべてのチャンクを1つの DataFrame に結合する
final_df = pd.concat(result_list, ignore_index=True)
print(f"最終的に合計 {len(final_df)} 行の有効データ")

pd.read_sql_query() と pd.read_sql_table()

pd.read_sql_query(): クエリの実行のみ

とpd.read_sql()機能はほぼ同じですが、SQL クエリ文のみを受け付け、テーブル名は受け付けません。パラメータは完全に同じで、「クエリ」と「テーブル読み取り」の操作を明確に区別したいシナリオに適しています。

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

df = pd.read_sql_query(
    "SELECT name, salary FROM employees WHERE salary > 5000",
    con=engine
)
print(df)

pd.read_sql_table(): テーブル全体を読み取る（SQLAlchemy のみ）

pd.read_sql_table()テーブル全体の読み取り専用で、パラメータによる列と行のフィルタリングをサポートしますが、SQLAlchemy エンジン接続のみをサポートし、sqlite3 などのネイティブ接続はサポートしません：

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# employees テーブルの指定した列を読み取る
df = pd.read_sql_table(
    "employees",
    con=engine,
    columns=["id", "name", "salary", "department"]  # これらの列のみ読み取る
)

# schema（データベース内のスキーマ）の指定もサポート
df = pd.read_sql_table(
    "employees",
    con=engine,
    schema="hr"    # hr スキーマの employees テーブルを読み取る
)

print(df.head())

DataFrame をデータベースに書き込む（to_sql）

使用DataFrame.to_sql()DataFrame をデータベースに書き込むことができます。新しいテーブルの作成、データの追加、または既存テーブルの上書きに対応しています：

基本構文

DataFrame.to_sql(name, con, schema=None, if_exists='fail', index=True, index_label=None, chunksize=None, dtype=None, method=None)

if_existsパラメータは、対象テーブルが既に存在する場合の動作を決定します：

if_exists の値 動作 適用シナリオ
'fail'（デフォルト） テーブルが既に存在する場合はエラーが発生します 既存データの誤上書きを防ぐ
'replace' 元のテーブルを削除してから再作成し、書き込む データを全量更新する
'append' 既存のテーブルにデータを追加し、テーブル構造は変更しない 新しいデータを増分で書き込む

例

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")

# サンプルデータを準備する
df = pd.DataFrame({
    "name": ["田中", "鈴木", "佐藤"],
    "department": ["IT", "HR", "IT"],
    "salary": [12000, 8000, 15000]
})

# DataFrame を employees テーブルに書き込む。既に存在する場合は置き換える
df.to_sql(
    "employees",
    con=engine,
    if_exists="replace",  # 元のデータを上書き
    index=False           # DataFrame の行インデックスをデータベースに書き込まない（通常は不要）
)

# 既存のテーブルに新しいデータを追加する（注意：列名とデータ型が一致している必要があります）
new_employees = pd.DataFrame({
    "name": ["高橋"],
    "department": ["Finance"],
    "salary": [11000]
})
new_employees.to_sql("employees", con=engine, if_exists="append", index=False)

# 書き込み結果を確認する
result = pd.read_sql("SELECT * FROM employees", con=engine)
print(result)

列のデータ型を指定する

書き込み時には、dtypeパラメータで各列のデータベース内の型を明示的に指定できます：

例

import pandas as pd
from sqlalchemy import create_engine, Integer, String, Float, DateTime

engine = create_engine("sqlite:///mydata.db")

df = pd.DataFrame({
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
    "score": [92.5, 88.0, 95.3],
})

df.to_sql(
    "students",
    con=engine,
    if_exists="replace",
    index=False,
    dtype={
        "id":    Integer(),   # 整数型
        "name":  String(50),  # 可変長文字列、最大50文字
        "score": Float()      # 浮動小数点数
    }
)

完全な使用例

以下は、SQLite データベースから売上データを読み取り、分析する完全な例です：

例

import pandas as pd
import sqlite3
from sqlalchemy import create_engine

# ステップ1：テストデータベースを準備する
engine = create_engine("sqlite:///sales.db")
conn = sqlite3.connect("sales.db")

# サンプルデータを作成してデータベースに書き込む
sales_data = pd.DataFrame({
    "order_id":   [1001, 1002, 1003, 1004, 1005, 1006],
    "product":    ["ノートパソコン", "スマートフォン", "タブレット", "ノートパソコン", "スマートフォン", "タブレット"],
    "amount":     [6999, 3999, 2999, 7299, 4299, 3199],
    "quantity":   [2, 5, 3, 1, 4, 2],
    "order_date": ["2024-01-10", "2024-01-15", "2024-01-20",
                   "2024-02-05", "2024-02-10", "2024-02-18"]
})
sales_data.to_sql("sales", con=engine, if_exists="replace", index=False)

# ステップ2：データを読み取って日付を解析する
df = pd.read_sql(
    "SELECT * FROM sales",
    con=engine,
    parse_dates=["order_date"]
)
print("元データ：")
print(df)
print()

# ステップ3：製品別に売上を集計する
sql_agg = """
    SELECT product,
COUNT(*) AS 注文数,
SUM(amount * quantity) AS 総売上高,
AVG(amount) AS 平均単価
    FROM sales
    GROUP BY product
ORDER BY 総売上高 DESC
"""
summary = pd.read_sql(sql_agg, con=engine)
print("製品別売上集計：")
print(summary)
print()

# ステップ4：集計結果をデータベースに書き戻す
summary.to_sql("sales_summary", con=engine, if_exists="replace", index=False)
print("集計データは sales_summary テーブルに書き込まれました")

conn.close()

上記のコードの実行結果は次のとおりです：

原始数据：
   order_id product  amount  quantity order_date
0      1001     笔记本    6999         2 2024-01-10
1      1002      手机    3999         5 2024-01-15
2      1003      平板    2999         3 2024-01-20
3      1004     笔记本    7299         1 2024-02-05
4      1005      手机    4299         4 2024-02-10
5      1006      平板    3199         2 2024-02-18

各产品销售汇总：
  product  订单数  总销售额   平均单价
0     手机    2   36991   4149.0
1    笔记本    2   21297   7149.0
2     平板    2   15395   3099.0

汇总数据已写入 sales_summary 表

よくある質問と注意点

1、接続は使用後に閉じる必要があります

DB-API ネイティブ接続（sqlite3 など）を使用する場合、操作完了後に接続を手動で閉じる必要があります。推奨されるのは、withステートメントによる自動管理：

例

import sqlite3
import pandas as pd

# with ステートメントを使用すると、終了時に自動的に接続が閉じられ、例外が発生しても接続がリークしません
with sqlite3.connect("mydata.db") as conn:
    df = pd.read_sql("SELECT * FROM employees", con=conn)

print(df)  # 接続は自動的に閉じられましたが、df データは引き続き使用できます

2、大きなテーブルでは、すべてのデータを直接読み取らないでください

数百万行以上の大きなテーブルでは、直接SELECT *全データを読み取るとメモリを使い果たします。SQL レベルでデータをフィルタリング（WHERE、LIMIT）するか、chunksizeチャンク読み取りを使用してください。

3、read_sql_table は SQLAlchemy のみをサポートします

sqlite3 などの DB-API ネイティブ接続で呼び出すとpd.read_sql_table()、エラーが発生します。NotImplementedError。代わりに使用してください。pd.read_sql()または SQLAlchemy エンジンに変更してください。

4、to_sql の index パラメータはデフォルトで True です

デフォルトではto_sqlDataFrame の行インデックス（0, 1, 2...）もデータベースに書き込まれ、名前がindexの列が作成されます。通常これは不要です。明示的に渡すことをお勧めしますindex=False。

5、データベースドライバは個別にインストールする必要があります

SQLAlchemy で異なるデータベースに接続する場合、対応するドライバパッケージのインストールも必要です：

データベース ドライバパッケージ インストールコマンド
MySQL PyMySQL pip install pymysql
PostgreSQL psycopg2 pip install psycopg2-binary
SQL Server pyodbc pip install pyodbc
Oracle cx_Oracle pip install cx_Oracle
SQLite 組み込み インストール不要
その他の拡張