Функция 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, ...)
Описание параметра
| Параметр | Type | Описание | Значение по умолчанию |
|---|---|---|---|
| sql | str или SQLAlchemy Selectable | SQL-запрос или имя таблицы | Обязательный |
| con | движок sqlalchemy или соединение sqlite3 | Объект соединения с базой данных | Обязательный |
| index_col | str, список str | Имя столбца, используемое в качестве индекса строк | Отсутствует |
| coerce_float | bool | Следует ли пытаться преобразовывать числовые строки в числа с плавающей запятой | True |
| params | list, tuple, dict | Параметры для параметризованных SQL-запросов | Отсутствует |
| parse_dates | list, dict | Столбцы, которые необходимо разобрать как даты | Отсутствует |
| chunksize | int | Размер итератора для чтения по частям | Отсутствует |
Возвращаемое значение
- Тип возвращаемого значения：
pd.DataFrameили Iterator[DataFrame] - Когда
chunksizeЕсли None, возвращается один 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-запрос в качестве первого параметра.conПараметр con требует передачи объекта подключения к базе данных, который может быть подключением sqlite3, SQLAlchemy и т. д.- Можно напрямую передать имя таблицы, чтобы прочитать данные всей таблицы.
index_colПараметр index_col может указать столбец в качестве индекса строк.
Пример 2: Подключение к базе данных с помощью SQLAlchemy
SQLAlchemy — самый популярный ORM-фреймворк в Python, поддерживающий подключения к нескольким базам данных.
Пример
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', 'Электроника', 999.99, 50),
('B', 'Электроника', 499.99, 100),
('C', 'Одежда', 29.99, 200),
('D', 'Продукты', 5.99, 500),
('E', 'Книги', 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(
запрос,
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', 'Электроника', 999.99, 50),
('B', 'Электроника', 499.99, 100),
('C', 'Одежда', 29.99, 200),
('D', 'Продукты', 5.99, 500),
('E', 'Книги', 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(
запрос,
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, и т. д.
paramsПараметр params используется для параметризованных запросов, что позволяет предотвратить атаки SQL-инъекций.- Использование SQLAlchemy
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("Тестовая база данных была очищена")
Ожидаемые результаты выполнения:
分块读取数据:
Chunk 1: 2000 行
块 读取 category='A' 的数据
...
块 5: 2000 行
该块 category='A' 的数量: 约 667 行
聚合结果:
category total
0 A 1687500
1 B 1678250
2 Chunks 读取 约 665 行
Пояснение к коду:
chunksizeПараметр chunksize возвращает итератор, и каждая итерация возвращает DataFrame.- Чтение по частям подходит для больших наборов данных и позволяет избежать проблем с нехваткой памяти.
- Для агрегационных запросов можно использовать
pd.concat()чтобы объединить результаты всех частей.
Примечания
- Чтобы использовать
read_sql()необходимо установить соответствующий драйвер базы данных. - SQLite не требует дополнительной установки и может использоваться напрямую.
- MySQL требует
pymysqlилиmysql-connector-python：pip install pymysql。 - PostgreSQL требует
psycopg2：pip install psycopg2。 - Использование параметризованных запросов (
paramsparameter) может предотвратить атаки SQL-инъекций. - При чтении больших таблиц используйте
chunksizeпараметр для чтения по частям.
Итог
read_sql()Это ключевая функция в pandas для подключения к базам данных и чтения данных. Она может выполнять любой SQL-запрос и преобразовывать результаты запроса в формат DataFrame.
В реальной работе по анализу данных базы данных являются основным способом хранения корпоративных данных. Уверенное владениеread_sql()позволяет напрямую получать данные из базы данных для анализа. Обратите внимание на использование параметризованных запросов для обеспечения безопасности и использование чтения по частям для обработки больших наборов данных.