用 MSSQL-Python 选择数据访问和分析模式

该mssql-python驱动程序提供多条路径用于读取 Microsoft SQL 的数据。 每条路径适合不同的工作量。 本指南帮助您根据数据量、分析需求和性能需求选择合适的方案。

按工作量决定

请使用这张表格找到你的起点：

工作量	建议的路径	为什么
应用行访问（Web API，CRUD）	游标提取方法	开销低，逐行处理，没有额外的依赖。
小到中型报告查询	pandas	熟悉的 API 用于过滤、分组和可视化。
大型结果集或宽表	箭头提取	零拷贝列式传输，内存开销极低。
高性能分析	极地人与箭侠	在列式数据上进行多线程执行，无 GIL 争用。
针对本地和远程数据的即席 SQL 查询	DuckDB 与 Arrow 合作	在 Arrow 表上进行 SQL 分析，并与本地 CSV/Parquet 文件联接。
笔记本探索	pandas 或 Polars with Arrow	根据团队熟悉度和数据量来选择。

游标提取方法

当你需要按行访问且无需额外依赖时，请使用标准光标方法。 对于逐行处理、返回 API 响应或输入应用逻辑的应用代码来说，这种方法是正确的选择。

import mssql_python

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

cursor = conn.cursor()

# fetchone(): Process rows one at a time
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ListPrice > %(threshold)s", {"threshold": 100})
row = cursor.fetchone()
while row:
    print(f"{row.Name}: ${row.ListPrice:.2f}")
    row = cursor.fetchone()

# fetchmany(): Process in batches
cursor.execute("SELECT ProductID, Name FROM Production.Product")
while True:
    batch = cursor.fetchmany(100)
    if not batch:
        break
    for row in batch:
        print(row.Name)

# fetchval(): Get a single scalar value
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()

使用 fetchmany() 以内存高效的方式对大型结果集进行批量处理。 当你需要单个值（如计数、最大值或存在性检查）时，请使用 fetchval()。

有关获取方法的完整文档，请参见 获取数据。

箭头提取

当您需要用于分析、构建 DataFrame 或导出到 Parquet 的列式数据时，请使用 Arrow 提取。 Arrow 提供从驱动程序进行零拷贝数据传输，从而避免了从 fetchall() 构建 DataFrame 时的逐行转换开销。

带有列存储索引的表在数据库引擎中本就以列式格式存储，因此提取为 Arrow 格式非常适合这类工作负载。

cursor.execute("""
    SELECT ProductID, Name, ListPrice, Color
    FROM Production.Product
    WHERE ListPrice > 0
""")

# Get a single Arrow table
arrow_table = cursor.arrow()
print(f"{arrow_table.num_rows} rows, {arrow_table.num_columns} columns")

对于大型结果集，可使用 arrow_reader() 以流式方式分批处理，而无需将所有内容都加载到内存中：

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")

# Stream Arrow record batches
reader = cursor.arrow_reader(batch_size=10000)
for batch in reader:
    # Each batch is a pyarrow.RecordBatch
    print(f"Batch: {batch.num_rows} rows")

箭头表是熊猫、极地和鸭子数据库的起点。 提取一次，然后转换：

cursor.execute("""
    SELECT ProductID, Name, ListPrice, Color
    FROM Production.Product
    WHERE ListPrice > 0
""")
arrow_table = cursor.arrow()

# Arrow -> pandas
df = arrow_table.to_pandas()

# Arrow -> Polars (zero-copy)
import polars as pl
df = pl.from_arrow(arrow_table)

完整的 Arrow 文档请参见 Apache Arrow 集成。

pandas

当你需要熟悉的DataFrame API进行报告、临时分析或数据清理时，可以使用Pandas。 pandas 在处理适合放入内存的结果集时效果最佳（最多可达几百万行，具体取决于列宽）。

cursor.execute("""
    SELECT p.Name, p.ListPrice, pc.Name AS Category
    FROM Production.Product p
    JOIN Production.ProductSubcategory ps ON p.ProductSubcategoryID = ps.ProductSubcategoryID
    JOIN Production.ProductCategory pc ON ps.ProductCategoryID = pc.ProductCategoryID
    WHERE p.ListPrice > 0
""")

import pandas as pd
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
df = pd.DataFrame.from_records(rows, columns=columns)

# Analyze
print(df.groupby("Category")["ListPrice"].agg(["mean", "count"]))

对于较大的结果集，使用 Arrow 构建 DataFrame，而不是使用 fetchall()：

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
arrow_table = cursor.arrow()
df = arrow_table.to_pandas()

关于完整的 PANDAS 模式，包括 ETL、时间序列和写回，请参见 pandas 集成。

极地人与箭侠

当你需要对较大的结果集进行更快的 DataFrame 操作时，请使用 Polars。 Polars 使用 Apache Arrow 作为其内存格式，因此从 cursor.arrow() 进行传输时是零拷贝的。 Polars 还会在多个线程上运行操作，从而避免 CPU 密集型转换中的 GIL 争用。

import polars as pl

cursor.execute("""
    SELECT ProductID, Name, ListPrice, Color
    FROM Production.Product
    WHERE ListPrice > 0
""")

arrow_table = cursor.arrow()
df = pl.from_arrow(arrow_table)

# Filter and aggregate
result = (
    df.filter(pl.col("ListPrice") > 100)
    .group_by("Color")
    .agg(pl.col("ListPrice").mean().alias("AvgPrice"))
    .sort("AvgPrice", descending=True)
)
print(result)

对于流式传输大型结果集：

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")

reader = cursor.arrow_reader(batch_size=50000)
frames = []
for batch in reader:
    frames.append(pl.from_arrow(batch))

df = pl.concat(frames)

完整的极地分布模式，请参见 极地积分。

DuckDB 与 Arrow 合作

当你需要对提取数据运行SQL分析、将服务器数据与本地CSV或Parquet文件连接，或导出结果成文件格式时，可以使用DuckDB。 DuckDB 可直接处理 Arrow 表，无需拷贝即可访问数据。

import duckdb

cursor.execute("""
    SELECT ProductID, Name, ListPrice, Color
    FROM Production.Product
    WHERE ListPrice > 0
""")

products = cursor.arrow()

# Run DuckDB SQL on the Arrow table
result = duckdb.sql("""
    SELECT Color, AVG(ListPrice) AS AvgPrice, COUNT(*) AS Count
    FROM products
    WHERE Color IS NOT NULL
    GROUP BY Color
    ORDER BY AvgPrice DESC
""")
print(result.fetchdf())

将服务器数据与本地文件连接：

cursor.execute("SELECT CustomerID, TerritoryID FROM Sales.Customer")
customers = cursor.arrow()

# Join with a local CSV file
result = duckdb.sql("""
    SELECT c.CustomerID, c.TerritoryID, l.Region
    FROM customers c
    JOIN read_csv_auto('regions.csv') l ON c.TerritoryID = l.TerritoryID
""")

导出至Parquet：

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

duckdb.sql("COPY orders TO 'orders.parquet' (FORMAT PARQUET)")

完整的DuckDB模式，请参见 DuckDB集成。

影响读取路径决策的 Microsoft SQL 功能

数据库引擎具有直接影响哪条读取路径最优的功能。 在选择方法时请考虑以下特点：

列存储索引

带有 列存储索引 的表以列格式存储数据。 箭头提取是这些表的自然切换方式，因为数据在引擎中已经是列式的。 如果你的分析查询扫描的是拥有数百万行的宽表，服务器端的非集群列存储索引结合客户端的Arrow提取，能提供最佳的端到端吞吐量。

索引视图

索引视图会预先计算并将汇总结果或联接结果存储在服务器上。 如果你的 Pandas 或 Polars 分析需要反复计算相同的聚合结果，可以考虑创建一个索引视图，改为查询该视图。 服务器会随着底层数据的变化自动维护该视图。

查询存储

查询存储 跟踪查询执行统计数据随时间变化。 用它来判断哪些查询的开销足够大，值得采用 Arrow 提取和本地 DataFrame 分析，而哪些查询直接通过游标读取即可。 如果一次查询的运行时间只有几毫秒，使用游标提取就没问题。 如果扫描数百万行，Arrow提取和本地分析可能会降低服务器负载。

智能查询处理

Microsoft SQL 的智能查询处理功能，如自适应连接、行商店的批处理模式和内存授权反馈，能够自动优化查询执行。 这些功能无论你选择哪条客户端阅读路径都能有效，但它们最有利于大型分析查询。 对于大多数工作负载，你无需调整提示或执行计划。

流式传输大型结果集

对于内存中无法容纳的结果集，可以使用流模式：

带有 fetchmany() 的基于游标的流式传输：

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
while True:
    batch = cursor.fetchmany(5000)
    if not batch:
        break
    for row in batch:
        print(row[0])  # Process each row

基于 Arrow 的流式传输到 Parquet：

import pyarrow.parquet as pq

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
reader = cursor.arrow_reader(batch_size=50000)
writer = None

for batch in reader:
    if writer is None:
        writer = pq.ParquetWriter("orders.parquet", batch.schema)
    writer.write_batch(batch)

if writer:
    writer.close()

应避免的反模式

反模式	问题	更好的方法
fetchall()，然后为大型表格选择 pd.DataFrame()	将所有行加载到内存中两次（第一次以元组形式，第二次以 DataFrame 形式）。	先使用cursor.arrow()，再使用arrow_table.to_pandas()。
将Arrow转换为pandas以筛选行数	会因完整复制整个 pandas 对象而浪费内存。	用 SQL（WHERE 子句）过滤，或者直接在 Arrow 表上使用 Polars/DuckDB。
SELECT * 当你需要三列时	从服务器传输不必要的数据。	只列出你需要的列。
构建一个数据帧来计算 COUNT(*)	服务器计算聚合的速度比 Python 快。	使用 SELECT COUNT(*) 和 fetchval()。
每次查询开启新连接	即使考虑到连接池的开销，创建连接的成本仍然很高。	在逻辑工作单元内重复使用连接。
链箭 -> 熊猫 -> 极地	每次转换都会复制数据。	直接转换为目标格式：Arrow -> Polars 或 Arrow -> pandas。

相关内容

• 检索数据
• Apache Arrow 集成
• Pandas整合
• 极星积分
• DuckDB 集成
• 性能优化
• 数据加载与移动模式