9.5 Pandas 与数据库 (SQLAlchemy)

9.5 Pandas 与数据库 (SQLAlchemy)

9.5 Pandas 与数据库 (SQLAlchemy)

Pandas 提供了强大的工具，可以轻松地与数据库进行交互，允许我们从数据库读取数据到 Pandas DataFrame 中，也可以将 DataFrame 中的数据写入数据库。 SQLAlchemy 是一个流行的 Python SQL 工具包和 ORM（对象关系映射器），它提供了灵活的方式来连接到各种数据库，并执行 SQL 操作。 Pandas 能够很好地与 SQLAlchemy 集成，使得数据科学家和分析师能够方便地在 Pandas 和数据库之间进行数据交换。

9.5.1 SQLAlchemy 简介

SQLAlchemy 不是一个数据库，而是一个 Python 库，它提供了一组工具，用于与各种数据库进行交互。它支持多种数据库后端，包括 PostgreSQL、MySQL、SQLite、Oracle 等。 SQLAlchemy 提供两种主要的方式来与数据库交互：

• SQLAlchemy Core: 提供了一个 SQL 表达式语言，允许你以 Pythonic 的方式构建 SQL 查询，而无需编写原始 SQL 字符串。

• SQLAlchemy ORM: 提供了一个更高级别的抽象，允许你将数据库表映射到 Python 类，并通过对象来操作数据库。

9.5.2 安装 SQLAlchemy

在使用 Pandas 与 SQLAlchemy 之前，需要确保已经安装了这两个库。可以使用 pip 安装：

pip install pandas sqlalchemy
​

还需要安装特定数据库的驱动程序。例如，如果要连接到 PostgreSQL 数据库，需要安装 psycopg2：

pip install psycopg2
​

对于MySQL，你需要安装 mysqlclient 或者 pymysql。

pip install mysqlclient
# 或者
pip install pymysql
​

9.5.3 连接到数据库

要使用 SQLAlchemy 连接到数据库，首先需要创建一个 SQLAlchemy Engine 对象。Engine 对象负责管理与数据库的连接。

from sqlalchemy import create_engine
# SQLite 数据库连接
engine = create_engine('sqlite:///my_database.db')
# PostgreSQL 数据库连接
# engine = create_engine('postgresql://user:password@host:port/database')
# MySQL 数据库连接 (使用 mysqlclient)
# engine = create_engine('mysql+mysqldb://user:password@host:port/database')
# MySQL 数据库连接 (使用 pymysql)
# engine = create_engine('mysql+pymysql://user:password@host:port/database')
​

上述代码创建了一个连接到 SQLite 数据库的 Engine 对象。连接字符串的格式取决于你使用的数据库类型。

9.5.4 从数据库读取数据到 Pandas DataFrame

Pandas 提供了 read_sql() 函数，可以方便地从数据库读取数据到 DataFrame 中。 read_sql() 函数接受一个 SQL 查询字符串或一个 SQLAlchemy Selectable 对象，以及一个 Engine 对象作为参数。

import pandas as pd
from sqlalchemy import create_engine
# 创建一个 SQLite 数据库连接
engine = create_engine('sqlite:///my_database.db')
# 创建一个示例表
with engine.connect() as connection:
    connection.execute('''
        CREATE TABLE IF NOT EXISTS employees (
            id INTEGER PRIMARY KEY,
            name VARCHAR(255),
            department VARCHAR(255),
            salary INTEGER
        )
    ''')
    connection.execute('''
        INSERT INTO employees (name, department, salary) VALUES
        ('Alice', 'Sales', 50000),
        ('Bob', 'Marketing', 60000),
        ('Charlie', 'Engineering', 70000)
    ''')
# 使用 SQL 查询读取数据
sql_query = "SELECT * FROM employees"
df = pd.read_sql(sql_query, engine)
print(df)
# 使用 SQLAlchemy Selectable 对象读取数据
from sqlalchemy import text
sql_query = text("SELECT * FROM employees WHERE salary > :salary")
df = pd.read_sql(sql_query, engine, params={"salary": 55000})
print(df)
​

代码解释：

1. 首先，我们创建了一个 SQLite 数据库连接。

2. 然后，我们创建了一个名为 employees 的示例表，并插入了一些数据。

3. 接着，我们使用 read_sql() 函数从 employees 表中读取所有数据，并将结果存储在 DataFrame df 中。

4. 最后，我们打印 DataFrame 的内容。

5. 我们还展示了如何使用 SQLAlchemy 的 text 函数来构造参数化的 SQL 查询，以避免 SQL 注入风险。

9.5.5 将 Pandas DataFrame 写入数据库

Pandas 提供了 to_sql() 方法，可以将 DataFrame 中的数据写入数据库。 to_sql() 方法接受表名、Engine 对象和一些其他参数，例如 if_exists 和 index。

import pandas as pd
from sqlalchemy import create_engine
# 创建一个 SQLite 数据库连接
engine = create_engine('sqlite:///my_database.db')
# 创建一个示例 DataFrame
data = {'name': ['David', 'Eve', 'Frank'],
        'department': ['HR', 'Finance', 'IT'],
        'salary': [80000, 90000, 100000]}
df = pd.DataFrame(data)
# 将 DataFrame 写入数据库
df.to_sql('employees', engine, if_exists='append', index=False)
# 验证数据是否已成功写入数据库
df_read = pd.read_sql("SELECT * FROM employees", engine)
print(df_read)
​

代码解释：

1. 首先，我们创建了一个 SQLite 数据库连接。

2. 然后，我们创建了一个示例 DataFrame df。

3. 接着，我们使用 to_sql() 方法将 DataFrame df 写入 employees 表中。 if_exists='append' 表示如果表已存在，则将数据追加到表中。 index=False 表示不将 DataFrame 的索引写入数据库。

4. 最后，我们使用 read_sql() 函数从 employees 表中读取所有数据，并将结果存储在 DataFrame df_read 中，以验证数据是否已成功写入数据库。

9.5.6 使用 SQLAlchemy ORM

除了使用 SQLAlchemy Core 之外，还可以使用 SQLAlchemy ORM 来与数据库交互。 ORM 允许你将数据库表映射到 Python 类，并通过对象来操作数据库。

import pandas as pd
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
# 创建一个 SQLite 数据库连接
engine = create_engine('sqlite:///my_database.db')
# 定义一个基类
Base = declarative_base()
# 定义 Employee 类，映射到 employees 表
class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department = Column(String)
    salary = Column(Integer)
    def __repr__(self):
       return "<Employee(name='%s', department='%s', salary='%s')>" % (self.name, self.department, self.salary)
# 创建表
Base.metadata.create_all(engine)
# 创建一个会话
Session = sessionmaker(bind=engine)
session = Session()
# 创建 Employee 对象并添加到会话
alice = Employee(name='Alice', department='Sales', salary=50000)
bob = Employee(name='Bob', department='Marketing', salary=60000)
session.add_all([alice, bob])
session.commit()
# 查询 Employee 对象
employees = session.query(Employee).all()
for employee in employees:
    print(employee)
# 将查询结果转换为 DataFrame
df = pd.read_sql(session.query(Employee).statement, session.bind)
print(df)
​

代码解释：

1. 首先，我们创建了一个 SQLite 数据库连接。

2. 然后，我们定义了一个 Employee 类，并将其映射到 employees 表。

3. 接着，我们使用 Base.metadata.create_all(engine) 创建表。

4. 然后，我们创建了一个会话，并创建了两个 Employee 对象，并将它们添加到会话中。

5. 接着，我们提交会话，将数据写入数据库。

6. 然后，我们使用 session.query(Employee).all() 查询所有 Employee 对象。

7. 最后，我们使用 pd.read_sql() 函数将查询结果转换为 DataFrame。

9.5.7 错误处理

在与数据库交互时，可能会遇到各种错误，例如连接错误、查询错误等。应该使用 try...except 块来捕获这些错误，并进行适当的处理。

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
try:
    # 创建一个 SQLite 数据库连接
    engine = create_engine('sqlite:///non_existent_database.db')
    # 使用 SQL 查询读取数据
    sql_query = "SELECT * FROM employees"
    df = pd.read_sql(sql_query, engine)
    print(df)
except SQLAlchemyError as e:
    print(f"An error occurred: {e}")
​

代码解释：

1. 我们尝试连接到一个不存在的数据库 non_existent_database.db。

2. 如果连接失败，会抛出一个 SQLAlchemyError 异常。

3. 我们使用 try...except 块捕获这个异常，并打印错误信息。

9.5.8 最佳实践

• 使用参数化查询: 为了避免 SQL 注入风险，应该始终使用参数化查询。

• 显式关闭连接: 在完成数据库操作后，应该显式关闭连接，以释放资源。 虽然 SQLAlchemy 会自动管理连接池，但在某些情况下，显式关闭连接仍然是一个好习惯。

• 使用上下文管理器: 可以使用 with 语句来自动管理连接的打开和关闭。

• 批量插入数据: 当需要插入大量数据时，应该使用批量插入操作，以提高性能。 to_sql() 方法本身就支持批量插入，但也可以使用 SQLAlchemy Core 或 ORM 来实现更高级的批量插入。

• 索引优化: 如果经常需要查询某个列，可以考虑在该列上创建索引，以提高查询性能。

9.5.9 总结

Pandas 与 SQLAlchemy 的集成使得数据科学家和分析师能够方便地在 Pandas DataFrame 和数据库之间进行数据交换。通过 read_sql() 函数，可以轻松地从数据库读取数据到 DataFrame 中。通过 to_sql() 方法，可以将 DataFrame 中的数据写入数据库。 SQLAlchemy ORM 提供了更高级别的抽象，允许你将数据库表映射到 Python 类，并通过对象来操作数据库。 在与数据库交互时，应该注意错误处理和最佳实践，以确保代码的健壮性和性能。

9.5.10 Mermaid 图示

下面是一个简单的 Mermaid 图示，展示了 Pandas 与 SQLAlchemy 之间的交互流程：

图示解释：

1. Pandas DataFrame 通过 read_sql() 函数发送 SQL 查询到 SQLAlchemy Engine。

2. SQLAlchemy Engine 负责与数据库建立连接，并执行 SQL 查询。

3. 数据库返回查询结果到 SQLAlchemy Engine。

4. SQLAlchemy Engine 将查询结果转换为 Pandas DataFrame。

5. Pandas DataFrame 通过 to_sql() 函数发送数据到 SQLAlchemy Engine。

6. SQLAlchemy Engine 负责将数据写入数据库。

希望这篇文章能够帮助你更好地理解 Pandas 与 SQLAlchemy 的集成。