Pandas lee bases de datos SQL

Pandas ofrece un conjunto de funciones que interactúan directamente con bases de datos SQL, permitiendo leer los resultados de consultas directamente como DataFrame y también escribir DataFrames de vuelta a la base de datos. Esto permite a los analistas de datos no tener que manejar manualmente la conexión a la base de datos y el análisis de resultados, simplificando en gran medida el flujo de interacción entre la base de datos y Python.

Resumen de funciones principales

Función	Propósito	Tipo de retorno
`pd.read_sql()`	Ejecuta consultas SQL o lee toda la tabla (función genérica compatible con ambos escenarios)	DataFrame
`pd.read_sql_query()`	Ejecuta sentencias de consulta SQL, adecuado para consultas complejas	DataFrame
`pd.read_sql_table()`	Lee directamente toda la tabla, solo compatible con conexiones SQLAlchemy	DataFrame
`DataFrame.to_sql()`	Escribe un DataFrame en una tabla de la base de datos	None / int

En la práctica, se recomienda usarpd.read_sql(), que automáticamente determina según los parámetros pasados si ejecutar una consulta o leer la tabla completa, con la mejor compatibilidad.pd.read_sql_table()Solo admite conexiones con motor SQLAlchemy, no admite conexiones nativassqlite3como DB-API.

Establecer conexión a la base de datos

Pandas en sí no se conecta directamente a la base de datos; es necesario establecer una conexión mediante una biblioteca de terceros y pasarla. Hay dos formas principales:Motor SQLAlchemy(recomendado) yconexión nativa DB-API(ligera y sencilla).

Método 1: SQLAlchemy (recomendado)

SQLAlchemy es el paquete de herramientas de bases de datos más utilizado en Python, compatible con todas las bases de datos principales y con la mejor compatibilidad con Pandas:

pip install sqlalchemy

Ejemplo

from sqlalchemy import create_engine

# Formato de cadena de conexión SQLAlchemy: tipo de base de datos+driver://usuario:contraseña@host:puerto/nombre_base_datos
# SQLite (base de datos de archivo, sin necesidad de usuario ni contraseña)
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")

# Verificar si la conexión fue exitosa
with engine.connect() as conn:
print("Conexión exitosa")

Método 2: conexión nativa sqlite3 (solo SQLite)

Módulo integrado de Pythonsqlite3, no requiere instalación adicional, ideal para bases de datos SQLite locales ligeras:

Ejemplo

import sqlite3

# Conectar a una base de datos SQLite de archivo (se crea automáticamente si el archivo no existe)
conn = sqlite3.connect("mydata.db")

# Conectar a una base de datos en memoria (los datos desaparecen al salir del programa, ideal para pruebas)
conn_memory = sqlite3.connect(":memory:")

# Después de su uso, es necesario cerrar la conexión manualmente
# conn.close()

pd.read_sql() leer datos

Sintaxis básica

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

Descripción de los parámetros principales:

Parámetro	Tipo	Descripción
`sql`	str	Sentencia SQL de consulta, o nombre de tabla (según el`con`tipo)
`con`	Objeto de conexión	Motor SQLAlchemy u objeto de conexión DB-API
`index_col`	str o list	Establecer las columnas especificadas como índice de filas del DataFrame
`params`	list o dict	Valores de parámetros para consultas SQL parametrizadas, previene la inyección SQL
`parse_dates`	list o dict	Analizar las columnas especificadas como tipo datetime
`chunksize`	int	Lectura por bloques; cada bloque devuelve un iterador de DataFrame con el número de filas especificado

1. Leer toda la tabla

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Leer toda la tabla employees
df = pd.read_sql("employees", con=engine)
print(df.head())
print(f"Total {len(df)} filas, {len(df.columns)} columnas")

2. Ejecutar consultas SQL

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Consulta con condiciones
df = pd.read_sql("SELECT * FROM employees WHERE department = 'IT'", con=engine)

# Consulta con unión de múltiples tablas
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. Consultas parametrizadas (prevención de inyección SQL)

Cuando las condiciones de la consulta provienen de la entrada del usuario,asegúrese de usar consultas parametrizadas, no use concatenación de cadenas SQL:

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# ❌ Escritura peligrosa: concatenación de cadenas SQL, con riesgo de inyección SQL
dept = "IT"
# df = pd.read_sql(f"SELECT * FROM employees WHERE department = '{dept}'", engine)

# ✅ Escritura segura: usar el marcador de posición ? (sqlite3) o el parámetro nombrado :name (SQLAlchemy)
# Estilo SQLite / DB-API (usando ?)
df = pd.read_sql(
"SELECT * FROM employees WHERE department = ? AND salary > ?",
con=engine,
params=["IT", 8000] # params es una lista, reemplaza ? por posición
)

# Estilo de parámetros nombrados SQLAlchemy (usando :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. Establecer la columna índice

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Establecer la columna id de la base de datos como índice de filas del DataFrame
df = pd.read_sql("SELECT * FROM employees", con=engine, index_col="id")
print(df.head())

# Usar múltiples columnas como índice compuesto
df = pd.read_sql(
"SELECT * FROM orders",
con=engine,
index_col=["year", "month"] # Índice compuesto
)

5. Analizar columnas de fecha

Los campos de fecha almacenados en la base de datos se leen por defecto como cadenas; usandoparse_datesel parámetro se pueden analizar directamente comodatetime：

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Analizar las columnas created_at y updated_at como tipo 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]

# También se puede especificar el formato de análisis (para formatos de fecha no estándar)
df = pd.read_sql(
"SELECT * FROM orders",
con=engine,
parse_dates={"created_at": "%Y%m%d"} # Analizar un formato como "20240115" como fecha
)

Lectura de grandes volúmenes de datos por bloques (chunksize)

Cuando la cantidad de datos en la tabla de la base de datos es muy grande, leer todo de una vez en memoria puede provocar OOM (desbordamiento de memoria). Usandochunksizeel parámetro se pueden leer los datos en lotes, procesando solo una parte cada vez:

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# chunksize=10000 indica que cada vez se leen 10000 filas, devolviendo un iterador
chunks = pd.read_sql("SELECT * FROM large_table", con=engine, chunksize=10000)

# Procesar bloque por bloque (solo hay 10000 filas en memoria cada vez)
result_list = []
for i, chunk in enumerate(chunks):
# Ejecutar lógica de procesamiento en cada bloque (como filtrado, agregación, etc.)
processed = chunk[chunk["status"] == "active"]
result_list.append(processed)
print(f"Procesado bloque {i+1}, filas válidas en el bloque actual: {len(processed)}")

# Combinar todos los bloques procesados en un solo DataFrame
final_df = pd.concat(result_list, ignore_index=True)
print(f"Finalmente un total de {len(final_df)} filas de datos válidos")

pd.read_sql_query() y pd.read_sql_table()

pd.read_sql_query(): solo ejecutar consultas

ypd.read_sql()La funcionalidad es básicamente la misma, pero solo acepta sentencias SQL de consulta, no nombres de tabla. Los parámetros son exactamente los mismos, adecuado para escenarios donde se necesita distinguir claramente entre operaciones de "consulta" y "lectura de tabla":

Ejemplo

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(): leer toda la tabla (solo SQLAlchemy)

pd.read_sql_table()Diseñado específicamente para leer tablas completas, admite filtrar columnas y filas mediante parámetros, perosolo admite conexiones con motor SQLAlchemy, no admite conexiones nativas como sqlite3:

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Leer columnas específicas de la tabla employees
df = pd.read_sql_table(
"employees",
con=engine,
columns=["id", "name", "salary", "department"] # Leer solo estas columnas
)

# También admite especificar schema (esquema en la base de datos)
df = pd.read_sql_table(
"employees",
con=engine,
schema="hr" # Leer la tabla employees en el esquema hr
)

print(df.head())

Escribir DataFrame en la base de datos (to_sql)

UsoDataFrame.to_sql()Puede escribir el DataFrame en la base de datos, con soporte para crear nuevas tablas, agregar datos o sobrescribir la tabla original:

Sintaxis básica

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

if_existsEl parámetro determina el comportamiento cuando la tabla de destino ya existe:

Valor de if_exists	Comportamiento	Escenario de uso
`'fail'`(por defecto)	Se produce un error si la tabla ya existe	Evita sobrescribir accidentalmente datos existentes
`'replace'`	Primero elimina la tabla original y luego la vuelve a crear para escribir los datos	Actualización completa de los datos
`'append'`	Agrega datos a la tabla existente sin cambiar la estructura de la tabla	Escritura incremental de nuevos datos

Ejemplo

import pandas as pd
from sqlalchemy import create_engine

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

# Preparar los datos de ejemplo
df = pd.DataFrame({
"name": ["Zhang San", "Li Si", Wang Wu],
"department": ["IT", "HR", "IT"],
"salary": [12000, 8000, 15000]
})

# Escribir el DataFrame en la tabla employees; si ya existe, reemplazarla
df.to_sql(
"employees",
con=engine,
if_exists="replace", # Sobrescribir los datos existentes
index=False # No escribir el índice de filas del DataFrame en la base de datos (normalmente no es necesario)
)

# Agregar nuevos datos a la tabla existente (nota: los nombres de columna y los tipos de datos deben coincidir)
new_employees = pd.DataFrame({
"name": ["Zhao Liu"],
"department": ["Finance"],
"salary": [11000]
})
new_employees.to_sql("employees", con=engine, if_exists="append", index=False)

# Verificar el resultado de la escritura
result = pd.read_sql("SELECT * FROM employees", con=engine)
print(result)

Especificar el tipo de dato de las columnas

Al escribir, se puede pasar el parámetrodtypepara especificar explícitamente el tipo de cada columna en la base de datos:

Ejemplo

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(), # Tipo entero
"name": String(50), # Cadena de longitud variable, con un máximo de 50 caracteres
"score": Float() # Número de punto flotante
}
)

Ejemplo de uso completo

A continuación se muestra un ejemplo completo de cómo leer datos de ventas desde una base de datos SQLite y analizarlos:

Ejemplo

import pandas as pd
import sqlite3
from sqlalchemy import create_engine

# Primer paso: preparar la base de datos de prueba
engine = create_engine("sqlite:///sales.db")
conn = sqlite3.connect("sales.db")

# Crear los datos de ejemplo y escribirlos en la base de datos
sales_data = pd.DataFrame({
"order_id": [1001, 1002, 1003, 1004, 1005, 1006],
"product": ["Portátil", "Móvil", "Tablet", "Portátil", "Móvil", "Tablet"],
"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)

# Segundo paso: leer los datos y analizar las fechas
df = pd.read_sql(
"SELECT * FROM sales",
con=engine,
parse_dates=["order_date"]
)
print("Datos originales:")
print(df)
print()

# Tercer paso: calcular las ventas por producto
sql_agg = """
SELECT product,
COUNT(*) AS num_pedidos,
SUM(amount * quantity) AS ventas_totales,
AVG(amount) AS precio_promedio
FROM sales
GROUP BY product
ORDER BY ventas_totales DESC
"""
summary = pd.read_sql(sql_agg, con=engine)
print("Resumen de ventas por producto:")
print(summary)
print()

# Cuarto paso: escribir los resultados estadísticos de vuelta en la base de datos
summary.to_sql("sales_summary", con=engine, if_exists="replace", index=False)
print("Los datos resumidos se han escrito en la tabla sales_summary")

conn.close()

El resultado de ejecutar el código anterior es:

原始数据：
   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 表

Problemas comunes y advertencias

1. Las conexiones deben cerrarse después de su uso

Al usar conexiones nativas DB-API (como sqlite3), la conexión debe cerrarse manualmente después de la operación. Se recomienda usar la declaraciónwithpara gestionarla automáticamente:

Ejemplo

import sqlite3
import pandas as pd

# Usar la declaración with: la conexión se cierra automáticamente al salir; incluso si se produce una excepción, no se filtrará la conexión
with sqlite3.connect("mydata.db") as conn:
df = pd.read_sql("SELECT * FROM employees", con=conn)

print(df) # La conexión se ha cerrado automáticamente, pero los datos de df siguen disponibles

2. No lea todos los datos de tablas grandes directamente

Para tablas grandes con más de varios millones de filas, utilizarSELECT *para leer todos los datos directamente agotará la memoria. Se debe filtrar los datos a nivel de SQL (WHERE, LIMIT) preferentemente, o utilizarchunksizela lectura por bloques.

3. read_sql_table solo admite SQLAlchemy

Si se utiliza una conexión nativa DB-API (como sqlite3) para llamar apd.read_sql_table(), se producirá el errorNotImplementedError. Por favor, utilicepd.read_sql()o cambie al motor SQLAlchemy.

4. El parámetro index de to_sql es True por defecto

Por defecto,to_sqltambién escribe el índice de filas del DataFrame (0, 1, 2...) en la base de datos, formando una columna llamadaindex, que generalmente es innecesaria. Se recomienda pasar explícitamenteindex=False。

5. El controlador de la base de datos debe instalarse por separado

Al conectar SQLAlchemy con diferentes bases de datos, también debe instalarse el paquete de controlador correspondiente:

Base de datos	Paquete de controlador	Comando de instalación
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	Integrado	No requiere instalación

Otras extensiones

Lectura y escritura de datos

Núcleo de procesamiento de datos

Manual de referencia

Estadística y casos

Pandas lee bases de datos SQL

Resumen de funciones principales

Establecer conexión a la base de datos

Método 1: SQLAlchemy (recomendado)

Ejemplo

Método 2: conexión nativa sqlite3 (solo SQLite)

Ejemplo

pd.read_sql() leer datos

Sintaxis básica

1. Leer toda la tabla

Ejemplo

2. Ejecutar consultas SQL

Ejemplo

3. Consultas parametrizadas (prevención de inyección SQL)

Ejemplo

4. Establecer la columna índice

Ejemplo

5. Analizar columnas de fecha

Ejemplo

Lectura de grandes volúmenes de datos por bloques (chunksize)

Ejemplo

pd.read_sql_query() y pd.read_sql_table()

pd.read_sql_query(): solo ejecutar consultas

Ejemplo

pd.read_sql_table(): leer toda la tabla (solo SQLAlchemy)

Ejemplo

Escribir DataFrame en la base de datos (to_sql)

Sintaxis básica

Ejemplo

Especificar el tipo de dato de las columnas

Ejemplo

Ejemplo de uso completo

Ejemplo

Problemas comunes y advertencias

Ejemplo