Pandas pd.read_sql() 関数
Pandas よく使う関数read_sql()pandas ライブラリにおいてデータベースからデータを読み取るための関数であり、SQLクエリとデータベーステーブルの直接読み取りをサポートします。
データベースはエンタープライズアプリケーションの中核コンポーネントであり、大量の業務データを格納しています。read_sql()さまざまな種類のデータベース（MySQL、PostgreSQL、SQLite など）に接続し、SQLクエリを実行して結果をDataFrameに変換できるため、その後のデータ分析と処理が容易になります。
基本構文とパラメータ
構文形式
pandas.read_sql(sql, con, index_col=None, coerce_float=True, params=None,
parse_dates=None, chunksize=None, dtype=None, ...)
パラメータの説明
| パラメータ | 型 | 説明 | デフォルト値 |
|---|---|---|---|
| sql | str または SQLAlchemy Selectable | SQLクエリ文またはテーブル名 | 必須 |
| con | sqlalchemy engine、または sqlite3 connection | データベース接続オブジェクト | 必須 |
| index_col | str, list of str | 行インデックスとして使用する列名 | None |
| coerce_float | bool | 数値文字列を浮動小数点数に変換するかどうか | True |
| params | list, tuple, dict | SQLパラメータ化クエリのパラメータ | None |
| parse_dates | list, dict | 日付として解析する必要がある列 | None |
| chunksize | int | 分割して返すイテレータのサイズ | None |
戻り値
- 戻り値の型：
pd.DataFrameまたは Iterator[DataFrame] - 当
chunksizeNone の場合は、単一の DataFrame を返します。 - 設定した
chunksize場合、イテレータを返し、反復ごとに DataFrame を返します。
例
以下の例を通じて、read_sql()のさまざまな使い方を完全に習得します。
例1：SQLite データベースを使用
SQLite は軽量な組み込み型データベースで、独立したデータベースサーバーが不要なため、学習やテストに非常に適しています。
例
import pandas as pd
import sqlite3
# インメモリ SQLite データベースを作成してテストデータを挿入
conn = sqlite3.connect(':memory:')
# テーブルを作成してデータを挿入
cursor = conn.cursor()
# 社員テーブルを作成
cursor.execute('''
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
city TEXT,
salary INTEGER
)
''')
# テストデータを挿入
employees_data = [
(1, 'Tom', 28, 'Beijing', 8000),
(2, 'Jerry', 35, 'Shanghai', 12000),
(3, 'Mike', 42, 'Guangzhou', 15000),
(4, 'Lucy', 26, 'Shenzhen', 7000),
(5, 'John', 31, 'Beijing', 9500)
]
cursor.executemany('INSERT INTO employees VALUES (?, ?, ?, ?, ?)', employees_data)
conn.commit()
# 例 1a: SQLクエリを実行してデータを読み取る
# sql: SQLクエリ文（必須）
# con: データベース接続オブジェクト（必須）
query = "SELECT name, age, city, salary FROM employees WHERE salary > 8000"
df_query = pd.read_sql(query, conn)
print("SQLクエリを実行:")
print(df_query)
print()
# 例 1b: テーブル全体を直接読み取る
# テーブル名を sql パラメータとして使用できる
df_table = pd.read_sql('employees', conn)
print("テーブル全体を読み取る:")
print(df_table)
print()
# 例 1c: index_col を使用してインデックスを設定
df_indexed = pd.read_sql('employees', conn, index_col='id')
print("インデックス列を設定:")
print(df_indexed)
import sqlite3
# インメモリ SQLite データベースを作成してテストデータを挿入
conn = sqlite3.connect(':memory:')
# テーブルを作成してデータを挿入
cursor = conn.cursor()
# 社員テーブルを作成
cursor.execute('''
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
city TEXT,
salary INTEGER
)
''')
# テストデータを挿入
employees_data = [
(1, 'Tom', 28, 'Beijing', 8000),
(2, 'Jerry', 35, 'Shanghai', 12000),
(3, 'Mike', 42, 'Guangzhou', 15000),
(4, 'Lucy', 26, 'Shenzhen', 7000),
(5, 'John', 31, 'Beijing', 9500)
]
cursor.executemany('INSERT INTO employees VALUES (?, ?, ?, ?, ?)', employees_data)
conn.commit()
# 例 1a: SQLクエリを実行してデータを読み取る
# sql: SQLクエリ文（必須）
# con: データベース接続オブジェクト（必須）
query = "SELECT name, age, city, salary FROM employees WHERE salary > 8000"
df_query = pd.read_sql(query, conn)
print("SQLクエリを実行:")
print(df_query)
print()
# 例 1b: テーブル全体を直接読み取る
# テーブル名を sql パラメータとして使用できる
df_table = pd.read_sql('employees', conn)
print("テーブル全体を読み取る:")
print(df_table)
print()
# 例 1c: index_col を使用してインデックスを設定
df_indexed = pd.read_sql('employees', conn, index_col='id')
print("インデックス列を設定:")
print(df_indexed)
実行結果の期待値:
执行 SQL 查询:
name age city salary
0 Tom 28 Beijing 8000
1 Jerry 35 Shanghai 12000
2 Mike 42 Guangzhou 15000
chunksize
空字符串 NaN NaN NaN NaN
3 Lucy 26 Shenzhen 7000
4 John 31 Beijing 9500
读取整个表:
id name age city salary
0 1 Tom 28 Beijing 8000
1 pandas 的 read_sql() 和 read_sql_query() 函数说明 read_sql() 是统一的接口，可以接受 SQL 语句或表名 read_sql_query() 只接受 SQL 语句，返回查询结果 read_sql_table() 只接受表名，返回整个表
2 3 Mike 42 Guangzhou 15000
3 4 Lucy 26 Shenzhen 7000
4 5 John 31 Beijing 9500
设置索引列:
name age city salary
id
1 Tom 28 Beijing 8000
2 Jerry 35 Shanghai 12000
3 Mike 42 Guangzhou 15000
4 Lucy 26 Shenzhen 7000
5 John 31 Beijing 9500
コードの解説:
read_sql()SQLクエリ文を最初のパラメータとして受け取ります。concon パラメータにはデータベース接続オブジェクトを渡す必要があります。sqlite3、SQLAlchemy などの接続を使用できます。- テーブル名を直接渡すことで、テーブル全体のデータを読み取ることができます。
index_colindex_col パラメータで任意の列を行インデックスとして指定できます。
例2：SQLAlchemy を使用してデータベースに接続
SQLAlchemy は Python で最も人気のある ORM フレームワークであり、複数のデータベース接続をサポートしています。
例
import pandas as pd
from sqlalchemy import create_engine, text
# SQLAlchemy を使用してインメモリ SQLite データベースを作成
engine = create_engine('sqlite:///:memory:')
# テストデータを作成
with engine.connect() as conn:
# 製品テーブルを作成
conn.execute(text('''
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER
)
'''))
# データを挿入
products_data = [
('A', 'Electronics', 999.99, 50),
('B', 'Electronics', 499.99, 100),
('C', 'Clothing', 29.99, 200),
('D', 'Food', 5.99, 500),
('E', 'Books', 19.99, 150)
]
conn.execute(text('''
INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)
'''), products_data)
conn.commit()
# 例 2a: SQLAlchemy エンジンを使用してデータを読み取る
query = "SELECT * FROM products WHERE category = :category"
df_params = pd.read_sql(
query,
engine,
params={'category': 'Electronics'} # パラメータ化クエリを使用して SQL インジェクションを防ぐ
)
print("パラメータ化クエリを使用:")
print(df_params)
print()
# 例 2b: テーブル名を直接使用して読み取る
df_all = pd.read_sql('products', engine)
print("テーブル全体を読み取る:")
print(df_all)
print()
# 例 2c: 他のデータベース接続を試す（MySQL を例に）
# 注意: 対応するデータベースドライバのインストールが必要
# pip install pymysql # MySQL
# pip install psycopg2 # PostgreSQL
# MySQL 接続例
# engine_mysql = create_engine('mysql+pymysql://username:password@host/database')
# df_mysql = pd.read_sql('SELECT * FROM table_name', engine_mysql)
print("注意: 実際に MySQL/PostgreSQL に接続するには対応するドライバのインストールが必要です")
from sqlalchemy import create_engine, text
# SQLAlchemy を使用してインメモリ SQLite データベースを作成
engine = create_engine('sqlite:///:memory:')
# テストデータを作成
with engine.connect() as conn:
# 製品テーブルを作成
conn.execute(text('''
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER
)
'''))
# データを挿入
products_data = [
('A', 'Electronics', 999.99, 50),
('B', 'Electronics', 499.99, 100),
('C', 'Clothing', 29.99, 200),
('D', 'Food', 5.99, 500),
('E', 'Books', 19.99, 150)
]
conn.execute(text('''
INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)
'''), products_data)
conn.commit()
# 例 2a: SQLAlchemy エンジンを使用してデータを読み取る
query = "SELECT * FROM products WHERE category = :category"
df_params = pd.read_sql(
query,
engine,
params={'category': 'Electronics'} # パラメータ化クエリを使用して SQL インジェクションを防ぐ
)
print("パラメータ化クエリを使用:")
print(df_params)
print()
# 例 2b: テーブル名を直接使用して読み取る
df_all = pd.read_sql('products', engine)
print("テーブル全体を読み取る:")
print(df_all)
print()
# 例 2c: 他のデータベース接続を試す（MySQL を例に）
# 注意: 対応するデータベースドライバのインストールが必要
# pip install pymysql # MySQL
# pip install psycopg2 # PostgreSQL
# MySQL 接続例
# engine_mysql = create_engine('mysql+pymysql://username:password@host/database')
# df_mysql = pd.read_sql('SELECT * FROM table_name', engine_mysql)
print("注意: 実際に MySQL/PostgreSQL に接続するには対応するドライバのインストールが必要です")
実行結果の期待値:
使用参数化查询: id name category price stock 0 1 A Electronics 999.99 50 的 read_sql() 配对 write_frame() 已弃用，使用 to_sql() 方法替代 to_sql() 可以将 DataFrame 写入数据库表 2 2 B Electronics 499.99 100 读取整个表: id 名称 类别 价格 库存 0 1 A Electronics 999.99 50 1 2 B ... 读取 499. 参数 100 2 3 C Clothing 29.99 200 3 4 配对 Food 5.99 500 4 5 E Books 19.99 150
コードの解説:
- SQLAlchemy は統一されたデータベース接続インターフェースを提供し、MySQL、PostgreSQL、Oracle などの複数のデータベースをサポートします。
paramsparams パラメータはパラメータ化クエリに使用され、SQL インジェクション攻撃を防ぐことができます。- SQLAlchemy の
create_engine()create_engine を使用してデータベースエンジンを作成します。
例3：大きなデータの分割読み込みと処理
クエリ結果のデータ量が非常に大きい場合は、分割読み込みを行うことでメモリを節約できます。
例
import pandas as pd
import sqlite3
# 大量のテストデータを作成
conn = sqlite3.connect('test_large.db')
cursor = conn.cursor()
# テーブルを作成
cursor.execute('''
CREATE TABLE large_table (
id INTEGER PRIMARY KEY,
value INTEGER,
category TEXT
)
''')
# 10000 件のテストデータを挿入
import random
large_data = [(i, random.randint(1, 1000), random.choice(['A', 'B', 'C']))
for i in range(1, 10001)]
cursor.executemany('INSERT INTO large_table VALUES (?, ?, ?)', large_data)
conn.commit()
# 例 3a: データを分割して読み取る
# chunksize パラメータは毎回返す行数を指定する
print("データを分割読み込み:")
chunks = pd.read_sql('SELECT * FROM large_table', conn, chunksize=2000)
# ジェネレータを使用して各チャンクを処理
for i, chunk in enumerate(chunks):
print(f"チャンク {i+1}: {len(chunk)} 行")
print(f" このチャンクの category='A' の数: {len(chunk[chunk['category'] == 'A'])}")
print()
# 例 3b: 集計計算（すべてのチャンクを結合）
# 全体の統計が必要な場合は、すべてのチャンクを結合できます
all_chunks = []
chunks = pd.read_sql('SELECT category, SUM(value) as total FROM large_table GROUP BY category',
conn, chunksize=1000)
for chunk in chunks:
all_chunks.append(chunk)
# 結合結果
result = pd.concat(all_chunks, ignore_index=True)
print("集計結果:")
print(result)
print()
# 接続を閉じる
conn.close()
# テストファイルをクリーンアップ
import os
os.remove('test_large.db')
print("テストデータベースはクリーンアップされました")
import sqlite3
# 大量のテストデータを作成
conn = sqlite3.connect('test_large.db')
cursor = conn.cursor()
# テーブルを作成
cursor.execute('''
CREATE TABLE large_table (
id INTEGER PRIMARY KEY,
value INTEGER,
category TEXT
)
''')
# 10000 件のテストデータを挿入
import random
large_data = [(i, random.randint(1, 1000), random.choice(['A', 'B', 'C']))
for i in range(1, 10001)]
cursor.executemany('INSERT INTO large_table VALUES (?, ?, ?)', large_data)
conn.commit()
# 例 3a: データを分割して読み取る
# chunksize パラメータは毎回返す行数を指定する
print("データを分割読み込み:")
chunks = pd.read_sql('SELECT * FROM large_table', conn, chunksize=2000)
# ジェネレータを使用して各チャンクを処理
for i, chunk in enumerate(chunks):
print(f"チャンク {i+1}: {len(chunk)} 行")
print(f" このチャンクの category='A' の数: {len(chunk[chunk['category'] == 'A'])}")
print()
# 例 3b: 集計計算（すべてのチャンクを結合）
# 全体の統計が必要な場合は、すべてのチャンクを結合できます
all_chunks = []
chunks = pd.read_sql('SELECT category, SUM(value) as total FROM large_table GROUP BY category',
conn, chunksize=1000)
for chunk in chunks:
all_chunks.append(chunk)
# 結合結果
result = pd.concat(all_chunks, ignore_index=True)
print("集計結果:")
print(result)
print()
# 接続を閉じる
conn.close()
# テストファイルをクリーンアップ
import os
os.remove('test_large.db')
print("テストデータベースはクリーンアップされました")
実行結果の期待値:
分块读取数据:
チャンク1: 2000 行
块 读取 category='A' 的数据
...
块 5: 2000 行
该块 category='A' 的数量: 约 667 行
聚合结果:
category total
0 A 1687500
1 B 1678250
2 Chunks 读取 约 665 行
コードの解説:
chunksizechunksize パラメータはイテレータを返し、反復ごとに DataFrame を返します。- 分割読み込みは大規模データセットに適しており、メモリ不足の問題を回避できます。
- 集計クエリの場合は、
pd.concat()pd.concat() を使用してすべてのチャンクの結果を結合できます。
注意事項
- 使用するには
read_sql()対応するデータベースドライバのインストールが必要です。 - SQLite は追加インストール不要で、そのまま使用できます。
- MySQL には
pymysql或mysql-connector-python：pip install pymysql。 - PostgreSQL には
psycopg2：pip install psycopg2。 - パラメータ化クエリの使用（
paramsパラメータ）は SQL インジェクション攻撃を防ぐことができます。 - 大きなテーブルを読み取る場合は、
chunksizechunksize パラメータを使用して分割読み込みします。
まとめ
read_sql()pandas でデータベースに接続してデータを読み取るための中心的な関数です。任意の SQL クエリを実行し、その結果を DataFrame 形式に変換できます。
実際のデータ分析業務では、データベースは企業データの中核的な保存方法であり、十分に習得すればread_sql()データベースから直接データを取得して分析を行うことができます。安全性を確保するためにパラメータ化クエリを使用し、大規模データセットを処理するために分割読み込みを使用することに注意してください。