Pandas & NumPy Academy · 课时

从 Pandas 运行 SQL 查询

Q: 「从 Pandas 运行 SQL 查询」这节课中我会学到什么？

使用 pd.read_sql_query 执行任意 SELECT 语句，并安全地参数化查询以避免 SQL 注入。 你通过在浏览器中直接运行的动手代码来练习 Pandas & NumPy Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

使用 pd.read_sql_query 执行任意 SELECT 语句，并安全地参数化查询以避免 SQL 注入。

从 Pandas 运行 SQL 查询 是 CoddyKit 上的免费 Pandas & NumPy Academy 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Pandas & NumPy Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Pandas & NumPy Academy 课程共包含 4 节课。

pd.read_sql：统一接口

Pandas 提供三个 SQL 读取函数：pd.read_sql()（通用封装器）、pd.read_sql_table()（按名称读取完整数据表）和 pd.read_sql_query()（执行任意 SQL）。对于大多数分析流程，pd.read_sql_query() 最为强大，因为它允许您在数据进入 Pandas 之前编写包含筛选、连接和聚合的任意 SELECT 语句。使用 SQL 完成繁重的数据处理，再用 Pandas 进行最终分析，通常比先加载全部数据再在 Python 中筛选更高效。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///ecommerce.db')

# Three equivalent patterns
df1 = pd.read_sql('SELECT * FROM orders LIMIT 100', con=engine)
df2 = pd.read_sql_table('orders', con=engine)  # full table
df3 = pd.read_sql_query('SELECT * FROM orders LIMIT 100', con=engine)

print(df3.head())
print(df3.columns.tolist())

在数据库层进行筛选

请始终在 SQL 中筛选数据，而不是先加载全部数据再在 Pandas 中筛选。具有适当索引的数据库可以在数百万行中执行 WHERE 子句，并在几毫秒内只返回几千行；而 Pandas 则需要先加载数 GB 的数据。黄金法则是：将谓词下推到数据库。使用 WHERE 筛选行，使用 SELECT col1, col2 选择列，并在开发过程中使用 LIMIT 快速预览结果。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Filter and project at SQL level — only fetch what you need
query = '''
    SELECT order_id, customer_id, amount, status
    FROM orders
    WHERE status = 'completed'
      AND order_date >= '2024-01-01'
      AND amount > 50
    LIMIT 1000
'''

df = pd.read_sql_query(query, con=engine)
print(f'Rows: {len(df)}, Columns: {list(df.columns)}')

在 SQL 与 Pandas 中进行聚合

对于大型数据表上的简单分组汇总，SQL 聚合的性能优于 Pandas，因为数据库引擎可以使用索引、并行执行和磁盘上的哈希聚合。当数据表很大时，请使用 SQL 执行 GROUP BY 和 SUM/COUNT/AVG。将聚合结果（一个较小的 DataFrame）加载到 Pandas 中，以便进一步分析、可视化或与其他数据结合。对于 SQL 无法表达的复杂自定义聚合，请先将筛选后的数据子集加载到 Pandas 中，再在那里使用分组操作。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Aggregate in SQL — returns a small result set
query = '''
    SELECT region,
           COUNT(*) AS order_count,
           ROUND(SUM(amount), 2) AS total_revenue,
           ROUND(AVG(amount), 2) AS avg_order_value
    FROM orders
    WHERE status = 'completed'
    GROUP BY region
    ORDER BY total_revenue DESC
'''

df = pd.read_sql_query(query, con=engine)
print(df)

SQL 查询中的 JOIN

在大型数据表上执行连接时，SQL 的 JOIN 操作比 Pandas 的 merge() 更高效，因为数据库可以使用索引查找。在 SQL 中编写连接，并在 Pandas 中接收已经连接且可能已筛选的结果。对于多表分析，使用包含多个 JOIN 的单个 SQL 查询，通常比分别读取每个数据表再在 Pandas 中合并更快，尤其是在其中一个数据表包含数百万行且连接操作能显著减少结果规模时。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///ecommerce.db')

query = '''
    SELECT o.order_id,
           c.customer_name,
           c.country,
           p.product_name,
           o.amount
    FROM orders o
    JOIN customers c ON o.customer_id = c.id
    JOIN products p ON o.product_id = p.id
    WHERE o.status = 'completed'
    LIMIT 500
'''

df = pd.read_sql_query(query, con=engine)
print(df.head())

在查询中使用 Python 变量

要安全地将 Python 变量传入 SQL 查询，请使用带命名参数的 SQLAlchemy 的 text()。在查询字符串中使用 :param_name 定义占位符，并将字典传递给 read_sql_query 的 params 参数。这既适用于单个值，也适用于列表（具体取决于数据库）。请避免使用 f 字符串或 % 格式化根据变量构建查询字符串；即使在内部使用，这些方式也不安全。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Python variables to inject
min_amount = 200.0
start_date = '2024-01-01'
end_date = '2024-12-31'

query = sa.text('''
    SELECT * FROM orders
    WHERE amount > :min_amount
      AND order_date BETWEEN :start_date AND :end_date
''')

with engine.connect() as conn:
    df = pd.read_sql_query(query, con=conn,
                           params={'min_amount': min_amount,
                                   'start_date': start_date,
                                   'end_date': end_date})
print(f'{len(df)} orders found')

使用 CTE 和子查询

复杂分析通常需要公用表表达式（CTE）或子查询。pd.read_sql_query 完全支持这些功能——只需将包含多个子句的完整 SQL 作为查询字符串传入即可。CTE 以 WITH 关键字开头，通过为中间结果命名，让复杂查询更易读。这对于计算累计值、在分组内排名，以及执行在 Pandas 中会显得冗长的多步筛选非常有用。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

query = '''
    WITH monthly_revenue AS (
        SELECT strftime('%Y-%m', order_date) AS month,
               SUM(amount) AS revenue
        FROM orders
        WHERE status = 'completed'
        GROUP BY month
    )
    SELECT month,
           revenue,
           revenue - LAG(revenue) OVER (ORDER BY month)
               AS month_over_month_change
    FROM monthly_revenue
    ORDER BY month
'''

df = pd.read_sql_query(query, con=engine)
print(df.tail())

使用 DatetimeIndex 读取数据

从数据库读取时间序列数据时，请将时间戳列设置为 DataFrame 的索引：向 read_sql_query 传入 index_col='date_column' 和 parse_dates=['date_column']。这样就能直接得到 DatetimeIndex，无需额外的后处理步骤即可使用 Pandas 的基于时间的切片（df['2024-01']）、重采样和滚动计算。parse_dates 参数会指示 Pandas 将该列转换为 datetime64。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///metrics.db')

df = pd.read_sql_query(
    'SELECT recorded_at, metric_value FROM daily_metrics ORDER BY recorded_at',
    con=engine,
    index_col='recorded_at',
    parse_dates=['recorded_at']
)
print(df.index.dtype)   # datetime64[ns]
print(df['2024-06'])    # Slice by month directly

分析查询速度缓慢的原因

当查询速度很慢时，请在 SELECT 前添加 SQL 关键字 EXPLAIN（在 SQLite 中使用 EXPLAIN QUERY PLAN），以查看数据库的执行计划。请查找您原本期望使用索引查找（'SEARCH TABLE'）却出现全表扫描（'SCAN TABLE'）的情况。在 WHERE 和 JOIN 列上缺少索引是查询速度慢的最常见原因。请在数据库中创建适当的索引，然后使用 EXPLAIN 再次检查，之后再重新运行 Pandas 数据处理流程。

import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Check if the query uses an index
with engine.connect() as conn:
    plan = conn.execute(sa.text(
        'EXPLAIN QUERY PLAN SELECT * FROM orders WHERE customer_id = 42'
    )).fetchall()
    for row in plan:
        print(row)
    # Look for 'SEARCH TABLE orders USING INDEX' — not 'SCAN TABLE'

大型结果集的分页

以交互方式遍历大型结果集时（例如，一次处理一页结果），请使用 SQL LIMIT 和 OFFSET 来实现分页。每次获取 N 行，处理完后再获取下一批 N 行。虽然这种方式不如 chunksize 方法高效（后者会维护游标），但当报告需要逐步显示行，或需要合并多个查询的结果时，分页非常有用。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')
page_size = 10000
offset = 0

while True:
    query = sa.text(
        'SELECT * FROM orders ORDER BY order_id LIMIT :limit OFFSET :offset'
    )
    with engine.connect() as conn:
        df = pd.read_sql_query(query, con=conn,
                               params={'limit': page_size, 'offset': offset})
    if len(df) == 0:
        break
    print(f'Page at offset {offset}: {len(df)} rows')
    offset += page_size

结合 SQL 查询与 Pandas 逻辑

最强大的模式是混合数据处理流程：使用 SQL 进行粗粒度筛选和聚合，然后使用 Pandas 执行 SQL 难以表达的细粒度转换（透视表、字符串解析、应用函数、滚动窗口）。从 SQL 中读取规模适中的结果集（几千行），然后对生成的 DataFrame 链式执行 Pandas 操作。这样既能结合两种工具的优势，又能让数据始终在同一个 Python 进程中流转。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# SQL: coarse filter and join
df = pd.read_sql_query('''
    SELECT o.customer_id, o.amount, o.order_date, c.country
    FROM orders o JOIN customers c ON o.customer_id = c.id
    WHERE o.status = 'completed'
''', con=engine, parse_dates=['order_date'])

# Pandas: rolling 30-day revenue per country
df = df.sort_values('order_date')
df['rolling_30d'] = (
    df.groupby('country')['amount']
    .transform(lambda x: x.rolling('30D').sum())
)
print(df.head())

数据库查询的错误处理

数据库查询可能会因网络超时、语法错误或连接丢失而失败。请将数据库调用包装在 try-except 块中：连接问题捕获 sqlalchemy.exc.OperationalError，SQL 语法错误捕获 sqlalchemy.exc.ProgrammingError。请记录包含上下文信息（查询、参数）的错误，然后采用指数退避重试，或优雅地处理失败。在生产数据处理流程中，区分暂时性错误（可重试）和永久性错误（需要修正 SQL）至关重要。

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

try:
    df = pd.read_sql_query(
        'SELECT * FROM nonexistent_table',
        con=engine
    )
except sa.exc.OperationalError as e:
    print(f'Connection or table error: {e}')
except sa.exc.ProgrammingError as e:
    print(f'SQL syntax error: {e}')
except Exception as e:
    print(f'Unexpected error: {type(e).__name__}: {e}')

快速检查

测试您对本课数据分析概念的理解。

课程回顾

本课中您学到了：pd.read_sql_query() 可以执行任意 SQL SELECT 并返回一个 DataFrame；将筛选和聚合操作下推到 SQL 中，比将完整表加载到 Pandas 更高效；而混合数据处理流程则结合 SQL 进行粗粒度数据缩减和 Pandas 进行细粒度自定义转换。接下来，我们将学习如何将 DataFrame 写回数据库表。

常见问题解答

「从 Pandas 运行 SQL 查询」课时是免费的吗？

是的 — 「从 Pandas 运行 SQL 查询」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Pandas & NumPy Academy 课程的其余内容，请升级到 CoddyKit PRO。 Pandas & NumPy Academy 课程共包含 4 节课。

「从 Pandas 运行 SQL 查询」这节课中我会学到什么？

使用 pd.read_sql_query 执行任意 SELECT 语句，并安全地参数化查询以避免 SQL 注入。你通过在浏览器中直接运行的动手代码来练习 Pandas & NumPy Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Pandas & NumPy Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Pandas & NumPy Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「从 Pandas 运行 SQL 查询」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Pandas & NumPy Academy 课中编写并运行代码吗？

能。每节 Pandas & NumPy Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Pandas & NumPy Academy