Función Pandas pd.read_sql()
Funciones comunes de Pandasread_sql()Es una función de la biblioteca pandas que se utiliza para leer datos de bases de datos, compatible con consultas SQL y la lectura directa de tablas de bases de datos.
Las bases de datos son componentes centrales de las aplicaciones empresariales, almacenando grandes cantidades de datos comerciales.read_sql()Puede conectarse a varios tipos de bases de datos (como MySQL, PostgreSQL, SQLite, etc.), ejecutar consultas SQL y convertir los resultados en un DataFrame, facilitando el posterior análisis y procesamiento de datos.
Sintaxis básica y parámetros
Formato de sintaxis
pandas.read_sql(sql, con, index_col=None, coerce_float=True, params=None,
parse_dates=None, chunksize=None, dtype=None, ...)
Descripción de parámetros
| Parámetro | Tipo | Descripción | Valor predeterminado |
|---|---|---|---|
| sql | str o SQLAlchemy Selectable | Sentencia de consulta SQL o nombre de tabla | Obligatorio |
| con | sqlalchemy engine, o conexión sqlite3 | Objeto de conexión a la base de datos | Obligatorio |
| index_col | str, list of str | Nombre de la columna utilizada como índice de filas | None |
| coerce_float | bool | Si se intenta convertir cadenas numéricas en números de punto flotante | True |
| params | list, tuple, dict | Parámetros de la consulta SQL parametrizada | None |
| parse_dates | list, dict | Columnas que deben analizarse como fechas | None |
| chunksize | int | Tamaño del iterador devuelto por bloques | None |
Valor de retorno
- Tipo de retorno：
pd.DataFrameo Iterator[DataFrame] - Cuando
chunksizeCuando es None, devuelve un único DataFrame. - Cuando se establece
chunksizedevuelve un iterador, cada iteración devuelve un DataFrame.
Ejemplos
A través de los siguientes ejemplos, domine completamenteread_sql()los diversos usos de.
Ejemplo 1: Uso de la base de datos SQLite
SQLite es una base de datos embebida y ligera que no requiere un servidor de base de datos independiente, muy adecuada para el aprendizaje y las pruebas.
Ejemplo
import pandas as pd
import sqlite3
# Crear una base de datos SQLite en memoria e insertar datos de prueba
conn = sqlite3.connect(':memory:')
# Crear la tabla e insertar datos
cursor = conn.cursor()
# Crear tabla de empleados
cursor.execute('''
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
city TEXT,
salary INTEGER
)
''')
# Insertar datos de prueba
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()
# Ejemplo 1a: Ejecutar consulta SQL y leer datos
# sql: Sentencia de consulta SQL (obligatorio)
# con: Objeto de conexión a la base de datos (obligatorio)
query = "SELECT name, age, city, salary FROM employees WHERE salary > 8000"
df_query = pd.read_sql(query, conn)
print("Ejecutar consulta SQL:")
print(df_query)
print()
# Ejemplo 1b: Leer toda la tabla directamente
# Se puede usar el nombre de la tabla como parámetro sql
df_table = pd.read_sql('employees', conn)
print("Leer toda la tabla:")
print(df_table)
print()
# Ejemplo 1c: Usar index_col para establecer el índice
df_indexed = pd.read_sql('employees', conn, index_col='id')
print("Establecer columna de índice:")
print(df_indexed)
import sqlite3
# Crear una base de datos SQLite en memoria e insertar datos de prueba
conn = sqlite3.connect(':memory:')
# Crear la tabla e insertar datos
cursor = conn.cursor()
# Crear tabla de empleados
cursor.execute('''
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
city TEXT,
salary INTEGER
)
''')
# Insertar datos de prueba
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()
# Ejemplo 1a: Ejecutar consulta SQL y leer datos
# sql: Sentencia de consulta SQL (obligatorio)
# con: Objeto de conexión a la base de datos (obligatorio)
query = "SELECT name, age, city, salary FROM employees WHERE salary > 8000"
df_query = pd.read_sql(query, conn)
print("Ejecutar consulta SQL:")
print(df_query)
print()
# Ejemplo 1b: Leer toda la tabla directamente
# Se puede usar el nombre de la tabla como parámetro sql
df_table = pd.read_sql('employees', conn)
print("Leer toda la tabla:")
print(df_table)
print()
# Ejemplo 1c: Usar index_col para establecer el índice
df_indexed = pd.read_sql('employees', conn, index_col='id')
print("Establecer columna de índice:")
print(df_indexed)
Resultado esperado de la ejecución:
执行 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
Análisis del código:
read_sql()Acepta una sentencia de consulta SQL como primer parámetro.conEl parámetro con requiere un objeto de conexión a la base de datos, que puede ser una conexión de sqlite3, SQLAlchemy, etc.- Se puede pasar directamente el nombre de la tabla para leer los datos de toda la tabla.
index_colEl parámetro index_col puede especificar una columna como índice de filas.
Ejemplo 2: Conexión a la base de datos con SQLAlchemy
SQLAlchemy es el framework ORM más popular en Python y admite la conexión a múltiples bases de datos.
Ejemplo
import pandas as pd
from sqlalchemy import create_engine, text
# Usar SQLAlchemy para crear una base de datos SQLite en memoria
engine = create_engine('sqlite:///:memory:')
# Crear datos de prueba
with engine.connect() as conn:
# Crear tabla de productos
conn.execute(text('''
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER
)
'''))
# Insertar datos
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()
# Ejemplo 2a: Usar el motor de SQLAlchemy para leer datos
query = "SELECT * FROM products WHERE category = :category"
df_params = pd.read_sql(
query,
engine,
params={'category': 'Electronics'} # Consulta parametrizada para prevenir la inyección SQL
)
print("Usar consulta parametrizada:")
print(df_params)
print()
# Ejemplo 2b: Leer directamente usando el nombre de la tabla
df_all = pd.read_sql('products', engine)
print("Leer toda la tabla:")
print(df_all)
print()
# Ejemplo 2c: Probar otras conexiones de base de datos (con MySQL como ejemplo)
# Nota: Se debe instalar el controlador de base de datos correspondiente
# pip install pymysql # MySQL
# pip install psycopg2 # PostgreSQL
# Ejemplo de conexión MySQL
# engine_mysql = create_engine('mysql+pymysql://username:password@host/database')
# df_mysql = pd.read_sql('SELECT * FROM table_name', engine_mysql)
print("Nota: Conectarse realmente a MySQL/PostgreSQL requiere instalar el controlador correspondiente")
from sqlalchemy import create_engine, text
# Usar SQLAlchemy para crear una base de datos SQLite en memoria
engine = create_engine('sqlite:///:memory:')
# Crear datos de prueba
with engine.connect() as conn:
# Crear tabla de productos
conn.execute(text('''
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER
)
'''))
# Insertar datos
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()
# Ejemplo 2a: Usar el motor de SQLAlchemy para leer datos
query = "SELECT * FROM products WHERE category = :category"
df_params = pd.read_sql(
query,
engine,
params={'category': 'Electronics'} # Consulta parametrizada para prevenir la inyección SQL
)
print("Usar consulta parametrizada:")
print(df_params)
print()
# Ejemplo 2b: Leer directamente usando el nombre de la tabla
df_all = pd.read_sql('products', engine)
print("Leer toda la tabla:")
print(df_all)
print()
# Ejemplo 2c: Probar otras conexiones de base de datos (con MySQL como ejemplo)
# Nota: Se debe instalar el controlador de base de datos correspondiente
# pip install pymysql # MySQL
# pip install psycopg2 # PostgreSQL
# Ejemplo de conexión MySQL
# engine_mysql = create_engine('mysql+pymysql://username:password@host/database')
# df_mysql = pd.read_sql('SELECT * FROM table_name', engine_mysql)
print("Nota: Conectarse realmente a MySQL/PostgreSQL requiere instalar el controlador correspondiente")
Resultado esperado de la ejecución:
使用参数化查询: 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
Análisis del código:
- SQLAlchemy proporciona una interfaz unificada de conexión a bases de datos, compatible con MySQL, PostgreSQL, Oracle y otras bases de datos.
paramsEl parámetro params se utiliza para consultas parametrizadas, lo que puede prevenir ataques de inyección SQL.- Usando la función de SQLAlchemy
create_engine()para crear el motor de la base de datos.
Ejemplo 3: Lectura por bloques y procesamiento de datos masivos
Cuando el volumen de datos del resultado de la consulta es muy grande, se puede leer por bloques para ahorrar memoria.
Ejemplo
import pandas as pd
import sqlite3
# Crear una gran cantidad de datos de prueba
conn = sqlite3.connect('test_large.db')
cursor = conn.cursor()
# Crear tabla
cursor.execute('''
CREATE TABLE large_table (
id INTEGER PRIMARY KEY,
value INTEGER,
category TEXT
)
''')
# Insertar 10000 registros de datos de prueba
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()
# Ejemplo 3a: Leer datos por bloques
# El parámetro chunksize especifica cuántas filas devolver cada vez
print("Leer datos por bloques:")
chunks = pd.read_sql('SELECT * FROM large_table', conn, chunksize=2000)
# Usar un generador para procesar cada bloque
for i, chunk in enumerate(chunks):
print(f"Bloque {i+1}: {len(chunk)} filas")
print(f" Cantidad de category='A' en este bloque: {len(chunk[chunk['category'] == 'A'])}")
print()
# Ejemplo 3b: Cálculo de agregación (combinar todos los bloques)
# Si se necesitan estadísticas globales, se pueden combinar todos los bloques
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)
# Combinar resultados
result = pd.concat(all_chunks, ignore_index=True)
print("Resultado de agregación:")
print(result)
print()
# Cerrar conexión
conn.close()
# Limpiar archivos de prueba
import os
os.remove('test_large.db')
print(La base de datos de prueba ha sido limpiada)
import sqlite3
# Crear una gran cantidad de datos de prueba
conn = sqlite3.connect('test_large.db')
cursor = conn.cursor()
# Crear tabla
cursor.execute('''
CREATE TABLE large_table (
id INTEGER PRIMARY KEY,
value INTEGER,
category TEXT
)
''')
# Insertar 10000 registros de datos de prueba
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()
# Ejemplo 3a: Leer datos por bloques
# El parámetro chunksize especifica cuántas filas devolver cada vez
print("Leer datos por bloques:")
chunks = pd.read_sql('SELECT * FROM large_table', conn, chunksize=2000)
# Usar un generador para procesar cada bloque
for i, chunk in enumerate(chunks):
print(f"Bloque {i+1}: {len(chunk)} filas")
print(f" Cantidad de category='A' en este bloque: {len(chunk[chunk['category'] == 'A'])}")
print()
# Ejemplo 3b: Cálculo de agregación (combinar todos los bloques)
# Si se necesitan estadísticas globales, se pueden combinar todos los bloques
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)
# Combinar resultados
result = pd.concat(all_chunks, ignore_index=True)
print("Resultado de agregación:")
print(result)
print()
# Cerrar conexión
conn.close()
# Limpiar archivos de prueba
import os
os.remove('test_large.db')
print(La base de datos de prueba ha sido limpiada)
Resultado esperado de la ejecución:
分块读取数据:
Fragmento 1: 2000 行
块 读取 category='A' 的数据
...
块 5: 2000 行
该块 category='A' 的数量: 约 667 行
聚合结果:
category total
0 A 1687500
1 B 1678250
2 Chunks 读取 约 665 行
Análisis del código:
chunksizeEl parámetro devuelve un iterador, y cada iteración devuelve un DataFrame.- La lectura por bloques es adecuada para conjuntos de datos grandes y puede evitar el problema de memoria insuficiente.
- Para consultas de agregación, se puede usar
pd.concat()para combinar los resultados de todos los bloques.
Notas
- Usar
read_sql()requiere instalar el controlador de base de datos correspondiente. - SQLite no requiere instalación adicional, se puede usar directamente.
- MySQL requiere
pymysqlomysql-connector-python：pip install pymysql。 - PostgreSQL requiere
psycopg2：pip install psycopg2。 - Usar consultas parametrizadas (
paramsparámetro) puede prevenir ataques de inyección SQL. - Al leer tablas grandes, usar el
chunksizeparámetro para lectura por bloques.
Resumen
read_sql()Es la función central de pandas para conectarse a bases de datos y leer datos. Puede ejecutar cualquier consulta SQL y convertir los resultados al formato DataFrame.
En el trabajo real de análisis de datos, las bases de datos son el método central de almacenamiento de datos empresariales. Dominarread_sql()permite obtener datos directamente de la base de datos para su análisis. Preste atención al uso de consultas parametrizadas para garantizar la seguridad y al uso de la lectura por bloques para procesar conjuntos de datos grandes.