mssql-python 應用程式的效能調校

mssql-python 驅動程式提供多項功能與模式以優化 SQL Server 應用程式效能，包括連線池、查詢優化及批量操作。

連線管理

使用連線池化

內建連線池。當你呼叫 conn.close() 時，連線會回到連線池中以供重複使用，而不是被銷毀，因此後續對 connect() 的呼叫會略過代價高昂的握手程序：

import mssql_python

def get_data():
    conn = mssql_python.connect(
        "Server=<server>.database.windows.net;Database=<database>;"
        "Authentication=ActiveDirectoryDefault;Encrypt=yes"
    )
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
        return cursor.fetchall()
    finally:
        conn.close()

設定工作負載的集區大小

根據你的並發需求調整池大小。如果你的應用程式同時處理多個使用者，請增加池數。對於較輕的工作負載，較小的池子能節省伺服器資源：

import mssql_python

mssql_python.pooling(
    max_size=50,      # Default is 100; reduce or increase for your workload
    idle_timeout=600  # Seconds before idle connections are recycled
)

在作業中重複使用連接

即使有池化，每個查詢都開啟新連線會增加負擔。相反地，在邏輯操作期間只保留一個連線：

# Bad: New connection per query
def bad_pattern(product_ids):
    for pid in product_ids:
        conn = mssql_python.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
        row = cursor.fetchone()
        process(row)
        conn.close()

# Good: Single connection for all queries
def good_pattern(product_ids):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    try:
        for pid in product_ids:
            cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
            row = cursor.fetchone()
            process(row)
    finally:
        conn.close()

在長期服務中保持連線暢通

網頁伺服器、佇列工作程序及持續執行的排程工作，應維持連線開啟，而非每次操作時都重新連線再中斷連線。開啟連線涉及 TCP 握手、TLS 協商及驗證，依網路距離與認證方式不同，可能需要 50-200 毫秒。對於每小時處理數千則訊息的佇列工作者來說，這種負擔會迅速累積。

在工作程序的整個生命週期內保持連線開啟，並在連線失敗時重新連線。在迭代間休息，以避免在隊列空時對伺服器造成重創：

import mssql_python
import time

def run_worker(connection_string: str, poll_interval: float = 1.0):
    conn = None
    try:
        while True:
            try:
                if conn is None:
                    conn = mssql_python.connect(connection_string)
                cursor = conn.cursor()
                cursor.execute("SELECT TOP 1 * FROM dbo.JobQueue WHERE Status = 'Pending' ORDER BY CreatedDate")
                job = cursor.fetchone()
                if job:
                    try:
                        process_job(job)
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Done' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    except Exception:
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Failed' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    conn.commit()
                else:
                    time.sleep(poll_interval)  # No work available, wait before polling again
            except mssql_python.OperationalError:
                # Connection lost, reconnect on next iteration
                conn = None
                time.sleep(poll_interval)
    finally:
        if conn is not None:
            conn.close()

啟用連線集區功能（預設）時，集區會為您處理閒置連線。但如果你停用連線集區，或使用單一專用連線，請在連線字串中設定 Connection Timeout 和 Command Timeout，以便及早偵測失效連線，而不是一直卡住。

查詢最佳化

僅擷取所需資料

只選擇應用程式所使用的欄位，可以減少網路傳輸、記憶體消耗及查詢執行時間。

# Bad: Select all columns
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})

# Good: Select specific columns
cursor.execute("""
    SELECT SalesOrderID, OrderDate, TotalDue
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(customer_id)s
""", {"customer_id": 1})

使用適當的擷取方法

驅動程式提供多種擷取方法。使用與你結果尺寸相符的版本：

fetchval() 回傳一個純量值，且開銷最小。
fetchall() 將整個結果集載入記憶體，這對於小型資料表效果很好。
fetchmany(n) 以批次方式檢索資料列，對於大型結果集保持記憶體使用不變。

正確的批次大小 fetchmany() 取決於行寬度。對於窄列（幾個小欄位，每個約 1 KB），1,000 列可維持每批約 1 MB 記憶體。對於含有大型字串或二進位欄位的較寬資料列，請使用較小的批次大小。從1,000開始，根據你的數據調整。

def process_batch(rows):
    # Example: print each row. Replace with your own logic.
    for row in rows:
        print(row)

# Single value
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()

# Small result set
cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor.fetchall()

# Large result set - process in batches
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader")
while True:
    batch = cursor.fetchmany(1000)
    if not batch:
        break
    process_batch(batch)

使用伺服器端分頁

不要擷取所有資料列後再在 Python 中進行切片，而是使用 OFFSET/FETCH NEXT 僅擷取所需的頁面。

def get_page(cursor, page: int, page_size: int = 50) -> list:
    """Get paginated results efficiently."""
    offset = (page - 1) * page_size

    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY ProductID
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})

    return cursor.fetchall()

使用SET NOCOUNT ON

預設情況下，SQL Server 會在每個 DML 語句後發送「受影響列」訊息。 SET NOCOUNT ON 抑制這些訊息並減少網路流量。這是會話層級設定，連線後設定一次，不要每次查詢都嵌入。

# Set once after connecting
cursor.execute("SET NOCOUNT ON")

# All subsequent statements on this connection skip the row-count message
cursor.execute(
    "INSERT INTO Log (Message) VALUES (%(message)s)",
    {"message": "Log entry"}
)

選擇合適的插入方式

驅動程式提供三種資料插入方式，每種方式適合不同的尺度：

方法	行數	原因為何
`execute()`	每次通話 1 列	用於單列操作，例如表單提交或 API 處理器，需要立即插入 ID。
`executemany()`	~10-1,000 行	採用逐欄參數綁定，吞吐量優於迴圈。將每一列以參數化語句形式傳送。
`bulkcopy()`	數百排以上	採用TDS大批量插入協定，效率遠高於逐列插入。最適合資料載入、遷移和批次處理。

欲了解更多細節與範例，請參閱資料載入與移動模式。

帶有 execute（）的單一插入

適用於只需執行一次的插入作業，且需要立即取得結果的情況。 Production.Product 有數個 NOT NULL 欄位且無預設值，因此插入時列出了所有欄位：

from datetime import datetime

cursor.execute(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (%(name)s, %(number)s, %(safety)s, %(reorder)s,
            %(cost)s, %(price)s, %(days)s, %(start)s)
    """,
    {
        "name": "Widget", "number": "WG-1001",
        "safety": 100, "reorder": 75,
        "cost": 12.50, "price": 19.99,
        "days": 1, "start": datetime(2024, 1, 1),
    },
)
conn.commit()

使用 executemany() 進行批次插入

executemany() 以欄位綁定參數並有效率地傳送。用它處理中等批量，而不是用迴圈呼叫 execute() 。請注意，executemany() 需要搭配元組清單的位置 ? 標記，而 execute() 同時支援 ? 和搭配字典的具名 %(name)s 參數。關於每種樣式的詳細資料，請參見參數化查詢。

rows = [
    ("Widget A", "WG-1001", 100, 75, 12.50, 19.99, 1, datetime(2024, 1, 1)),
    ("Widget B", "WG-1002", 100, 75, 15.00, 24.99, 1, datetime(2024, 1, 1)),
    ("Widget C", "WG-1003", 100, 75, 18.00, 29.99, 1, datetime(2024, 1, 1)),
]

cursor.executemany(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """,
    rows,
)
conn.commit()

大批量複製

當吞吐量比每列控制更重要時，切換為 bulkcopy()。它透過 TDS 的大量插入協定串流資料列，避免參數化陳述式的每列額外負荷。 bulkcopy() 表現優於 executemany() 的確切臨界點取決於資料列寬度和網路延遲，但通常落在一兩百列左右。對於非常小的批次， executemany() 會比較簡單，因為 bulkcopy() 會建立獨立的內部連線並自動提交。

與 execute() 和 executemany()不同，是 bulkcopy() 依位置將值映射到欄位，而非欄位 INSERT 列表。傳入 column_mappings 以指定要載入的目標欄位名稱，讓來源元組對應到正確的欄位，而不是資料表開頭的識別欄：

result = cursor.bulkcopy(
    "Production.Product",
    rows,
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
)
print(f"Copied {result['rows_copied']} rows")

對於非常大的載入，請使用產生器避免將整個資料集載入記憶體，並設定 batch_size 為定期提交：

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy(
    "Production.Product",
    csv_rows("products.csv"),
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
    batch_size=5000,
)

快取策略

對於很少變動的參考資料（類別、查詢表、設定），請在應用程式中快取結果，而不是每次請求都查詢。

Python 的 functools.lru_cache 提供簡單的記憶化功能，但它會一直快取，直到程序重新啟動。如果底層資料可能會變更，請使用 cachetools.TTLCache 在時間限制後自動重新整理：

from cachetools import TTLCache, cached

category_cache = TTLCache(maxsize=1, ttl=300)  # Refresh every 5 minutes

@cached(category_cache)
def get_categories(connection_string: str) -> tuple:
    conn = mssql_python.connect(connection_string)
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
        return cursor.fetchall()
    finally:
        conn.close()

網路最佳化

盡量減少往返次數

每次查詢都需要與伺服器進行一次網路往返。將相關查詢合併為單一批次，並使用 nextset() 逐一瀏覽結果集：

# Bad: Multiple round trips
cursor.execute("SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
customer = cursor.fetchone()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
orders = cursor.fetchall()
cursor.execute("SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s)", {"customer_id": 1})
detail_count = cursor.fetchval()

# Good: Single round trip
cursor.execute("""
    SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s;
    SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s;
    SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s);
""", {"customer_id": 1})

customer = cursor.fetchone()
cursor.nextset()
orders = cursor.fetchall()
cursor.nextset()
detail_count = cursor.fetchval()

使用伺服器端處理複雜邏輯

把聚合和過濾推送到 SQL Server，而不是直接抓取原始資料列再用 Python 處理。伺服器回傳的是一個摘要列，而非可能有數千個細節列：

cursor.execute("""
    SELECT p.Name, COUNT(sod.SalesOrderDetailID) AS OrderCount, SUM(sod.LineTotal) AS TotalSales
    FROM Production.Product p
    JOIN Sales.SalesOrderDetail sod ON p.ProductID = sod.ProductID
    WHERE p.ProductID = %(product_id)s
    GROUP BY p.Name
""", {"product_id": 707})

避免交錯游標操作

mssql-python 驅動程式不支援多重主動結果集（MARS）。每個連線只能有一個游標有主動查詢。在執行下一次查詢前，先完整取得第一個結果集，或使用第二個連線：

connection_string = (
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

# Option 1: Fetch first, then query (single connection)
# Warning: This is an N+1 pattern. Each iteration is a round trip.
# Use this only when the JOIN in Option 2 is not possible.
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT ProductID FROM Production.Product WHERE ProductSubcategoryID = 1")
product_ids = [row[0] for row in cursor.fetchall()]

for pid in product_ids:
    cursor.execute("SELECT ProductID, LocationID, Quantity FROM Production.ProductInventory WHERE ProductID = %(product_id)s", {"product_id": pid})
    inventory = cursor.fetchone()

conn.close()

# Option 2: Use a JOIN instead of N+1 queries (preferred)
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
    SELECT p.ProductID, p.Name, i.Quantity
    FROM Production.Product p
    LEFT JOIN Production.ProductInventory i ON p.ProductID = i.ProductID
    WHERE p.ProductSubcategoryID = 1
""")
results = cursor.fetchall()
conn.close()

記憶體管理

將大型結果分段處理

將數百萬列的資料表載入清單，會消耗與整個結果集成正比的記憶體。使用 OFFSET 和 FETCH NEXT 對資料進行伺服器端分頁，並一次處理一個資料區塊。

def quote_id(identifier: str) -> str:
    """Quote a possibly schema-qualified SQL identifier to prevent SQL injection."""
    return ".".join("[" + part.replace("]", "]]") + "]" for part in identifier.split("."))

def process_large_table(cursor, table: str, columns: list[str], key_column: str, processor, chunk_size: int = 10000):
    """Process large table without loading all data."""
    safe_table = quote_id(table)
    safe_key = quote_id(key_column)
    col_list = ", ".join(quote_id(c) for c in columns)
    cursor.execute(f"SELECT COUNT(*) FROM {safe_table}")
    total = cursor.fetchval()

    offset = 0
    while offset < total:
        cursor.execute(f"""
            SELECT {col_list} FROM {safe_table}
            ORDER BY {safe_key}
            OFFSET ? ROWS
            FETCH NEXT ? ROWS ONLY
        """, (offset, chunk_size))

        chunk = cursor.fetchall()
        processor(chunk)

        offset += chunk_size
        print(f"Processed {min(offset, total)}/{total}")

# key_column must be unique, otherwise rows can be duplicated or skipped across pages
process_large_table(
    cursor,
    "Production.TransactionHistory",
    ["TransactionID", "ProductID", "Quantity", "ActualCost"],
    "TransactionID",
    lambda chunk: None,  # replace with your row-processing logic
)

使用產生器進行串流

Python 產生器包裝fetchmany()會讓記憶體使用量保持不變，不論資料表大小。呼叫者逐列迭代，且不載入完整結果集。如果來源特別大，請使用 UNION ALL 合併資料表，並以相同方式串流傳送合併後的結果。

def stream_query(cursor, query: str, params: dict = None, batch_size: int = 1000):
    cursor.execute(query, params or {})
    
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        for row in batch:
            yield row

# Union the live and archive transaction tables into one extra-large result set
query = """
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistory
    UNION ALL
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistoryArchive
"""

count = 0
for row in stream_query(cursor, query, batch_size=5000):
    count += 1
print(f"Streamed {count} rows")

及時清理資源

未封閉的連線會佔用伺服器資源，並可能耗盡連線池。使用上下文管理器，即使發生例外也能保證清理。

from contextlib import contextmanager

@contextmanager
def database_connection(connection_string: str):
    conn = mssql_python.connect(connection_string)
    try:
        yield conn
    finally:
        conn.close()

with database_connection(connection_string) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
    data = cursor.fetchall()

監視記憶體使用量

大型結果集、長壽命快取與連線物件都會消耗記憶體。如果你的應用程式以服務方式執行，來自未關閉游標或無界限快取的記憶體洩漏，最終可能導致該行程被作業系統或容器執行環境終止。

使用 Python 的tracemalloc模組來快照記憶體並找出最大配置。

import tracemalloc

tracemalloc.start()

# ... run your workload ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
    print(stat)

常見的意外記憶增長來源包括：

對一個回傳數百萬列的查詢呼叫 fetchall()。請改用 fetchmany() 或產生器。
未使用 maxsize 或 TTL 的查詢結果快取快取會持續成長，直到程序重新啟動。
在迴圈中建立游標而不關閉它們。每個開啟的游標都會將其結果集保留在記憶體中。

索引與查詢計畫優化

檢查伺服器端查詢效能

用 SET STATISTICS TIME ON 和 SET STATISTICS IO ON 來查看查詢在伺服器上需要多長時間，以及它們讀取了多少資料。邏輯讀取次數偏高通常意味著缺少索引。請在 SQL Server Management Studio 或 Visual Studio Code 的 MSSQL 擴充功能中執行這些語句，輸出會顯示在訊息面板中：

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

您應該會看到如下的輸出：

Table 'Product'. Scan count 1, logical reads 3
SQL Server Execution Times: CPU time = 0 ms, elapsed time = 1 ms.

如果發現邏輯讀取次數偏高或出現資料表掃描，請考慮新增索引。

使用查詢提示作為戰術上的解決方法。

查詢提示會覆蓋查詢優化器對索引與聯結策略的選擇。在正式環境中，當查詢效能突然倒退時，它們是快速且低風險的修補方式，很有價值。你可以立即在應用程式程式碼中部署提示，穩定查詢，同時調查根本原因（缺少索引、統計過時或結構變更）。

避免讓暗示永久存在。當資料分布或結構改變時，硬編碼的提示可能會讓情況更糟。將它們視為暫時性，待根本問題解決後再回訪：

cursor.execute("""
    SELECT * FROM Production.Product WITH (INDEX(AK_Product_Name))
    WHERE ProductSubcategoryID = %(subcategory_id)s
""", {"subcategory_id": 1})

使用 OPTION（重新編譯）來繞過不良快取計畫

SQL Server 會根據它看到的第一組參數值來快取查詢計畫。若資料分布在呼叫間差異很大，快取計畫在某些值下可能表現不佳。這個稱為參數嗅探的問題，通常表現為原本很快的查詢突然要花上幾秒甚至幾分鐘。

OPTION (RECOMPILE)SQL Server 強制每次執行都建立新的計畫，這是個有效且即時的修復方法，你可以部署而無需伺服器端的任何變更。這種取捨的代價是每次呼叫都會有少量的編譯成本，但對於執行頻率不高或回傳大小可變結果集的查詢而言，與執行不佳的執行計畫相比，這點成本幾乎可以忽略不計。

一旦問題穩定，你可以慢慢進行永久性修正，例如重寫查詢、新增篩選索引，或使用計畫指南：

cursor.execute("""
    SELECT * FROM Sales.SalesOrderHeader
    WHERE OrderDate > %(start_date)s
    OPTION (RECOMPILE)
""", {"start_date": start_date})

效能監控

請計時你的查詢時間

要找到慢速運算，請將查詢包裝為：time.perf_counter()

import time

start = time.perf_counter()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": customer_id})
rows = cursor.fetchall()
elapsed = time.perf_counter() - start

print(f"Query returned {len(rows)} rows in {elapsed:.3f}s")

若要更全面地了解應用程式將時間花在哪些地方，請使用 Python 內建的 cProfile 模組：

python -m cProfile -s cumtime my_app.py

此視圖顯示每個函式呼叫的累積時間，有助於判斷緩慢是出在查詢執行、資料處理或網路延遲。

使用查詢存放區進行伺服器端分析

用戶端時序告訴你從應用程式角度查詢所需時間，但它結合了網路延遲、伺服器執行時間和用戶端處理。查詢存放區會擷取伺服器上的執行計畫和執行時統計，讓你能精確看到 SQL Server 如何執行每個查詢、執行頻率，以及其效能隨時間的變化。

查詢存放區特別適合辨識參數嗅探、計畫迴歸，以及消耗最多伺服器資源的查詢。你可以直接查詢 sys.query_store_runtime_stats 和 sys.query_store_plan 檢視，或使用 SQL Server Management Studio 內建的查詢存放區報告。

使用效能儀表板報表

SQL Server Management Studio 中的效能儀表板報告提供 SQL Server 健康狀況的即時總覽，包括目前等待類型、活躍且昂貴的查詢，以及 CPU/IO 趨勢。利用它們快速發現瓶頸，而不必直接向 DMV 查詢。

效能檢查清單

Connection

[ ] 啟用連線集區。
[ ] 根據您的工作負載調整集區大小。
[ ] 在作業中重用連線。
[ ] 在長期服務中保持聯繫暢通。

Queries

[ ] 只選擇你需要的欄位。
[ ] 為每個查詢使用適當的擷取方法。
[ ] 實作伺服器端分頁。
[ ] 連線後，設定 SET NOCOUNT ON 一次。
[ ] 透過批次查詢來減少往返次數。

Inserts

[ ] 單排插入時使用 execute()。
[ ] 針對小到中型批次（約 10 到 1,000 列），請使用 executemany()。
[ ] 當吞吐量比每列控制更重要時使用 bulkcopy() 。

Caching

[ ] 用 TTL 快取參考資料以避免提供過時的結果。

Resources

[ ] 以區塊或產生器處理大型結果。
[ ] 及時清理連接。
[ ] 使用 tracemalloc 監控長時間執行之服務的記憶體使用量。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22