Pandas SQL-Datenbank lesen

Pandas bietet eine Reihe von Funktionen für die direkte Interaktion mit SQL-Datenbanken. Abfrageergebnisse können direkt in einen DataFrame eingelesen werden, und ein DataFrame kann auch in die Datenbank zurückgeschrieben werden. Dadurch können Datenanalysten die manuelle Handhabung von Datenbankverbindungen und das Parsen von Ergebnissen vermeiden, was die Interaktion zwischen Datenbanken und Python erheblich vereinfacht.

Überblick über die Kernfunktionen

Funktion	Zweck	Rückgabetyp
`pd.read_sql()`	Führt SQL-Abfragen aus oder liest eine gesamte Tabelle (eine Allzweckfunktion, die mit beiden Szenarien kompatibel ist)	DataFrame
`pd.read_sql_query()`	Führt SQL-Abfrageanweisungen aus, geeignet für komplexe Abfragen	DataFrame
`pd.read_sql_table()`	Liest direkt eine gesamte Tabelle, unterstützt nur SQLAlchemy-Verbindungen	DataFrame
`DataFrame.to_sql()`	Schreibt ein DataFrame in eine Datenbanktabelle	None / int

In der praktischen Arbeit wird die Verwendung empfohlen,pd.read_sql()weil es anhand der übergebenen Parameter automatisch bestimmt, ob eine Abfrage ausgeführt oder eine gesamte Tabelle gelesen wird, und die beste Kompatibilität bietet.pd.read_sql_table()Unterstützt nur SQLAlchemy-Engine-Verbindungen; es unterstützt keine nativesqlite3und andere DB-API-Verbindungen.

Herstellen einer Datenbankverbindung

Pandas selbst stellt keine direkte Verbindung zu Datenbanken her. Es benötigt eine Drittanbieter-Bibliothek, um eine Verbindung herzustellen, die dann übergeben wird. Es gibt zwei gängige Ansätze:SQLAlchemy-Engine(empfohlen) undnative DB-API-Verbindung(leichtgewichtig und einfach).

Methode 1: SQLAlchemy (empfohlen)

SQLAlchemy ist Pythons am weitesten verbreitetes Datenbank-Toolkit. Es unterstützt alle wichtigen Datenbanken und hat die beste Kompatibilität mit Pandas:

pip install sqlalchemy

Beispiel

from sqlalchemy import create_engine

# SQLAlchemy-Verbindungsstring-Format: Datenbanktyp + driver://username:password@host:port/database name
# SQLite (dateibasierte Datenbank, kein Benutzername oder Passwort erforderlich)
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")

# Überprüfen, ob die Verbindung erfolgreich ist
with engine.connect() as conn:
print("Verbindung erfolgreich")

Methode 2: Native sqlite3-Verbindung (nur SQLite)

In Python integriertessqlite3Modul erfordert keine zusätzliche Installation und eignet sich für leichte lokale SQLite-Datenbanken:

Beispiel

import sqlite3

# Verbindung zu einer SQLite-Dateidatenbank herstellen (wird automatisch erstellt, wenn die Datei nicht existiert)
conn = sqlite3.connect("mydata.db")

# Verbindung zu einer In-Memory-Datenbank herstellen (Daten verschwinden nach Programmende; geeignet für Tests)
conn_memory = sqlite3.connect(":memory:")

# Die Verbindung muss nach der Verwendung manuell geschlossen werden
# conn.close()

pd.read_sql() Daten lesen

Grundlegende Syntax

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

Beschreibung der Hauptparameter:

Parameter	Type	Beschreibung
`sql`	str	SQL-Abfrageanweisung oder Tabellenname (abhängig vom`con`typ)
`con`	Verbindungsobjekt	SQLAlchemy-Engine oder DB-API-Verbindungsobjekt
`index_col`	str oder list	Legt die angegebene Spalte als Zeilenindex des DataFrame fest
`params`	list oder dict	Parameterwerte für parametrisierte SQL-Abfragen, die SQL-Injection verhindern
`parse_dates`	list oder dict	Parst die angegebene Spalte als datetime-Typ
`chunksize`	int	Liest in Blöcken ein und gibt einen Iterator von DataFrames mit der angegebenen Anzahl von Zeilen pro Block zurück

1. Eine gesamte Tabelle lesen

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# Lies die gesamte employees-Tabelle
df = pd.read_sql("employees", con=engine)
print(df.head())
print(f"Gesamt {len(df)} Zeilen, {len(df.columns)} Spalten")

2. Ausführen von SQL-Abfragen

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

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

# Abfrage mit Join über mehrere Tabellen
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. Parametrisierte Abfragen (Verhinderung von SQL-Injection)

Wenn Abfragebedingungen aus Benutzereingaben stammen,verwenden Sie immer parametrisierte Abfragenund verwenden Sie keine Zeichenkettenverkettung für SQL:

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# ❌ Gefährlicher Ansatz: Zeichenkettenverkettung von SQL, Risiko einer SQL-Injection
dept = "IT"
# df = pd.read_sql(f"SELECT * FROM employees WHERE department = '{dept}'", engine)

# ✅ Sichere Vorgehensweise: Verwenden Sie ?-Platzhalter (sqlite3) oder benannte Parameter :name (SQLAlchemy)
# SQLite / DB-API-Stil (mit ?)
df = pd.read_sql(
"SELECT * FROM employees WHERE department = ? AND salary > ?",
con=engine,
params=["IT", 8000] # params ist eine Liste, die ? positionsbasiert ersetzt
)

# SQLAlchemy-Stil für benannte Parameter (mit :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. Festlegen der Indexspalte

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# Die id-Spalte in der Datenbank als Zeilenindex des DataFrame festlegen
df = pd.read_sql("SELECT * FROM employees", con=engine, index_col="id")
print(df.head())

# Mehrere Spalten als zusammengesetzten Index verwenden
df = pd.read_sql(
"SELECT * FROM orders",
con=engine,
index_col=["year", "month"] # Zusammengesetzter Index
)

5. Parsen von Datumsspalten

In der Datenbank gespeicherte Datumsfelder werden standardmäßig als Zeichenfolgen gelesen. Mit demparse_datesParameter können sie direkt indatetime：

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# Die Spalten created_at und updated_at als datetime-Typ parsen
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]

# Sie können auch das Parsing-Format angeben (für nicht standardmäßige Datumsformate)
df = pd.read_sql(
"SELECT * FROM orders",
con=engine,
parse_dates={"created_at": "%Y%m%d"} # Ein Format wie "20240115" als Datum parsen
)

Große Datenmengen in Blöcken lesen (chunksize)

Wenn eine Datenbanktabelle eine sehr große Datenmenge enthält, kann das einmalige Einlesen aller Daten in den Speicher OOM (out of memory) verursachen. Mithilfe deschunksize-Parameters können Daten stapelweise gelesen werden, wobei jeweils nur ein Teil auf einmal verarbeitet wird:

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# chunksize=10000 bedeutet, dass jeweils 10000 Zeilen auf einmal gelesen werden und ein Iterator zurückgegeben wird
chunks = pd.read_sql("SELECT * FROM large_table", con=engine, chunksize=10000)

# Block für Block verarbeiten (nur 10000 Zeilen gleichzeitig im Speicher)
result_list = []
for i, chunk in enumerate(chunks):
# Wende Verarbeitungslogik auf jeden Chunk an (z. B. Filterung, Aggregation usw.)
processed = chunk[chunk["status"] == "active"]
result_list.append(processed)
print(f"Verarbeiteter Chunk {i+1}, gültige Zeilen im aktuellen Chunk: {len(processed)}")

# Kombiniere alle verarbeiteten Chunks zu einem einzigen DataFrame
final_df = pd.concat(result_list, ignore_index=True)
print(f"Endgültige gültige Datenzeilen: {len(final_df)}")

pd.read_sql_query() und pd.read_sql_table()

pd.read_sql_query(): Führt nur Abfragen aus

undpd.read_sql()Die Funktionalität ist grundsätzlich dieselbe, aber sie akzeptiert nur SQL-Abfrageanweisungen, keine Tabellennamen. Die Parameter sind genau dieselben, geeignet für Szenarien, in denen Sie klar zwischen "Abfrage"- und "Tabellenlesen"-Operationen unterscheiden müssen:

Beispiel

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(): Liest eine gesamte Tabelle (nur SQLAlchemy)

pd.read_sql_table()Wird speziell zum Lesen einer gesamten Tabelle verwendet, mit Unterstützung für das Filtern von Spalten und Zeilen über Parameter, aberunterstützt nur SQLAlchemy-Engine-Verbindungenund unterstützt keine nativen Verbindungen wie sqlite3:

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

# Angegebene Spalten aus der Tabelle employees lesen
df = pd.read_sql_table(
"employees",
con=engine,
columns=["id", "name", "salary", "department"] # Nur diese Spalten lesen
)

# Unterstützt auch die Angabe von schema (das Schema in der Datenbank)
df = pd.read_sql_table(
"employees",
con=engine,
schema="hr" # Die Tabelle employees unter dem Schema hr lesen
)

print(df.head())

Schreiben eines DataFrame in eine Datenbank (to_sql)

NutzungDataFrame.to_sql()Sie können einen DataFrame in eine Datenbank schreiben, mit Unterstützung für das Erstellen einer neuen Tabelle, das Anhängen von Daten oder das Überschreiben der ursprünglichen Tabelle:

Grundlegende Syntax

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

if_existsDer Parameter bestimmt das Verhalten, wenn die Zieltabelle bereits existiert:

if_exists-Wert	Verhalten	Anwendungsfall
`'fail'`(Standard)	Löst einen Fehler aus, wenn die Tabelle bereits existiert	Verhindert das versehentliche Überschreiben vorhandener Daten
`'replace'`	Löscht zuerst die ursprüngliche Tabelle, erstellt die Tabelle dann neu und schreibt in sie	Vollständige Datenaktualisierung
`'append'`	Hängt Daten an eine vorhandene Tabelle an, ohne die Tabellenstruktur zu ändern	Schreibt neue Daten inkrementell

Beispiel

import pandas as pd
from sqlalchemy import create_engine

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

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

# Schreibe das DataFrame in die Tabelle employees; ersetze sie, wenn sie bereits existiert
df.to_sql(
"employees",
con=engine,
if_exists="replace", # Ursprüngliche Daten überschreiben
index=False # Den Zeilenindex des DataFrame nicht in die Datenbank schreiben (normalerweise nicht erforderlich)
)

# Neue Daten an die vorhandene Tabelle anhängen (Hinweis: Spaltennamen und Datentypen müssen übereinstimmen)
new_employees = pd.DataFrame({
"name": ["Zhao Liu"],
"department": ["Finance"],
"salary": [11000]
})
new_employees.to_sql("employees", con=engine, if_exists="append", index=False)

# Schreibergebnis überprüfen
result = pd.read_sql("SELECT * FROM employees", con=engine)
print(result)

Spaltendatentypen angeben

Beim Schreiben können Sie dendtypeParameter verwenden, um den Typ jeder Spalte in der Datenbank explizit anzugeben:

Beispiel

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(), # Integer-Typ
"name": String(50), # Zeichenkette variabler Länge, maximal 50 Zeichen
"score": Float() # Gleitkommazahl
}
)

Vollständiges Anwendungsbeispiel

Nachfolgend finden Sie ein vollständiges Beispiel, wie man Verkaufsdaten aus einer SQLite-Datenbank liest und analysiert:

Beispiel

import pandas as pd
import sqlite3
from sqlalchemy import create_engine

# Schritt 1: Testdatenbank vorbereiten
engine = create_engine("sqlite:///sales.db")
conn = sqlite3.connect("sales.db")

# Beispieldaten erstellen und in die Datenbank schreiben
sales_data = pd.DataFrame({
"order_id": [1001, 1002, 1003, 1004, 1005, 1006],
"product": ["Laptop", "Phone", "Tablet", "Laptop", "Phone", "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)

# Schritt 2: Daten lesen und Datumsangaben parsen
df = pd.read_sql(
"SELECT * FROM sales",
con=engine,
parse_dates=["order_date"]
)
print("Rohdaten:")
print(df)
print()

# Schritt 3: Umsatz nach Produkt berechnen
sql_agg = """
SELECT product,
COUNT(*) AS order_count,
SUM(amount * quantity) AS total_sales,
AVG(amount) AS avg_unit_price
FROM sales
GROUP BY product
ORDER BY total_sales DESC
"""
summary = pd.read_sql(sql_agg, con=engine)
print("Verkaufsübersicht nach Produkt:")
print(summary)
print()

# Schritt 4: Die Zusammenfassungsergebnisse zurück in die Datenbank schreiben
summary.to_sql("sales_summary", con=engine, if_exists="replace", index=False)
print("Zusammenfassungsdaten wurden in die Tabelle sales_summary geschrieben")

conn.close()

Das Ausführungsergebnis des obigen Codes ist:

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

Häufige Probleme und Hinweise

1. Die Verbindung muss nach der Verwendung geschlossen werden

Bei der Verwendung einer nativen DB-API-Verbindung (wie sqlite3) müssen Sie die Verbindung nach dem Vorgang manuell schließen. Es wird empfohlen, diewithAnweisung zu verwenden, um sie automatisch zu verwalten:

Beispiel

import sqlite3
import pandas as pd

# Verwenden Sie die with-Anweisung; die Verbindung wird beim Verlassen automatisch geschlossen und selbst wenn eine Ausnahme auftritt, kommt es nicht zu einem Leck
with sqlite3.connect("mydata.db") as conn:
df = pd.read_sql("SELECT * FROM employees", con=conn)

print(df) # Die Verbindung wurde automatisch geschlossen, aber die df-Daten sind weiterhin verfügbar

2. Lesen Sie nicht alle Daten direkt aus einer großen Tabelle

Bei großen Tabellen mit Millionen von Zeilen oder mehr wird das direkteSELECT *Lesen aller Daten den Speicher erschöpfen. Sie sollten das Filtern von Daten auf SQL-Ebene (WHERE, LIMIT) priorisieren oderchunksizeblockweises Lesen verwenden.

3. read_sql_table unterstützt nur SQLAlchemy

Wenn Sie zum Aufrufen eine native DB-API-Verbindung wie sqlite3 verwendenpd.read_sql_table(), wird ein Fehler ausgelöstNotImplementedError. Bitte verwenden Siepd.read_sql()oder wechseln Sie zu einer SQLAlchemy-Engine.

4. Der index-Parameter von to_sql ist standardmäßig True

Standardmäßigto_sqlder Zeilenindex des DataFrame (0, 1, 2...) wird ebenfalls in die Datenbank geschrieben, wodurch eine Spalte namensindexSpalte erstellt wird, was normalerweise redundant ist. Es wird empfohlen, explizit zu übergebenindex=False。

5. Datenbanktreiber müssen separat installiert werden

Wenn SQLAlchemy eine Verbindung zu verschiedenen Datenbanken herstellt, müssen Sie außerdem das entsprechende Treiberpaket installieren:

Datenbank	Treiberpaket	Installationsbefehl
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	Integriert	Keine Installation erforderlich

Weitere Erweiterungen

Daten lesen und schreiben

Kern der Datenverarbeitung

Referenzhandbuch

Statistiken und Fälle

Pandas SQL-Datenbank lesen

Überblick über die Kernfunktionen

Herstellen einer Datenbankverbindung

Methode 1: SQLAlchemy (empfohlen)

Beispiel

Methode 2: Native sqlite3-Verbindung (nur SQLite)

Beispiel

pd.read_sql() Daten lesen

Grundlegende Syntax

1. Eine gesamte Tabelle lesen

Beispiel

2. Ausführen von SQL-Abfragen

Beispiel

3. Parametrisierte Abfragen (Verhinderung von SQL-Injection)

Beispiel

4. Festlegen der Indexspalte

Beispiel

5. Parsen von Datumsspalten

Beispiel

Große Datenmengen in Blöcken lesen (chunksize)

Beispiel

pd.read_sql_query() und pd.read_sql_table()

pd.read_sql_query(): Führt nur Abfragen aus

Beispiel

pd.read_sql_table(): Liest eine gesamte Tabelle (nur SQLAlchemy)

Beispiel

Schreiben eines DataFrame in eine Datenbank (to_sql)

Grundlegende Syntax

Beispiel

Spaltendatentypen angeben

Beispiel

Vollständiges Anwendungsbeispiel

Beispiel

Häufige Probleme und Hinweise

Beispiel