Polars 是一個以 Rust 撰寫的高效能 DataFrame 函式庫，提供一種快速且節省記憶體的 panda 替代方案。 將 Polars 與 mssql-python 驅動程式搭配使用，可讓你：
- 直接將 SQL 查詢結果載入 Polars DataFrames。
- 使用 Apache Arrow 從 Microsoft SQL 進行零複製資料傳輸。
- 有效率地將 Polars DataFrames 寫回 Microsoft SQL。
- 用懶惰的評估建立高效能的資料管線。
本文中的範例會查詢 AdventureWorks 範例資料庫。 如果你還沒有，請參考 AdventureWorks 的範例資料庫。
將資料讀取至 Polars DataFrame
你可以用兩種方式載入 Microsoft SQL 資料到 Polars：透過標準游標方法逐列轉換，或透過 Apache Arrow 零複製傳輸。 「零複製」表示資料會停留在單一的記憶體緩衝區，驅動程式、Arrow 和 Polars 都能直接讀取，因此不會有列重複成中間的 Python 物件。 大多數工作負載都採用 Arrow 方法，因為這種效率很高。
對 DataFrame 的基本查詢
此方法以標準游標擷取所有列，並手動建立 Polars 資料框架。 它接受參數 化查詢 以進行安全值替換。 它不需要 PyArrow 也能運作，但對於大型結果集會比較慢，因為每個值都會經過 Python。
import polars as pl
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
def query_to_polars(cursor, query: str, params: dict = None) -> pl.DataFrame:
"""Execute query and return results as Polars DataFrame."""
cursor.execute(query, params or {})
columns = [col[0] for col in cursor.description]
rows = cursor.fetchall()
data = {col: [row[i] for row in rows] for i, col in enumerate(columns)}
return pl.DataFrame(data)
# Usage: %(color)s is a parameterized placeholder. The driver safely substitutes
# the value from the dict, which prevents SQL injection.
df = query_to_polars(cursor, "SELECT TOP 5 Name, ListPrice FROM Production.Product WHERE Color = %(color)s", {"color": "Black"})
print(df)
Note
如果你的 連接字串 使用Authentication=ActiveDirectoryDefault，驅動程式會使用 DefaultAzureCredential，並依序嘗試多個憑證提供者。 第一次連線可能會比較慢，因為 SDK 會一直走鏈條直到找到可用的供應商。 在生產環境中，如果您知道您的環境使用哪一種憑證類型，請直接指定該憑證類型（例如，針對受控識別可指定 ActiveDirectoryMSI），以避免逐一嘗試整個憑證鏈。 如需詳細資訊，請參閱 Microsoft Entra 驗證。
建議使用 Arrow 進行零拷貝傳輸
將 Microsoft SQL 資料載入 Polars 最有效率的方式是透過 Apache Arrow。 mssql-python 驅動程式的 arrow() 方法會傳回一個 pyarrow.Table，可供 Polars 以零複製開銷取用。
def query_to_polars_arrow(cursor, query: str, params: dict = None) -> pl.DataFrame:
"""Execute query and load results through Arrow for best performance."""
cursor.execute(query, params or {})
arrow_table = cursor.arrow()
return pl.from_arrow(arrow_table)
# Usage
df = query_to_polars_arrow(cursor, "SELECT ProductID, Name, ListPrice FROM Production.Product")
print(df)
使用 Arrow 批次串流大型資料集
對於無法放入記憶體的資料集，請使用 arrow_reader() 串流批次處理資料。 每一批都是一個 pyarrow.RecordBatch，Polars 可以獨立處理，因此記憶體用量會與 batch_size 成正比，而不是與整個結果集成正比。 使用讀取器作為情境管理器，即使有例外中斷迴圈，也能釋放伺服器端游標：
def process_large_query(cursor, query: str, params: dict = None, batch_size: int = 50000) -> pl.DataFrame:
"""Process large query results as streaming Arrow batches."""
cursor.execute(query, params or {})
results = []
with cursor.arrow_reader(batch_size=batch_size) as reader:
for batch in reader:
chunk_df = pl.from_arrow(batch)
# Process each chunk
results.append(chunk_df)
return pl.concat(results) if results else pl.DataFrame()
# Usage
df = process_large_query(cursor, "SELECT * FROM Production.TransactionHistory")
使用 LazyFrames 進行延遲執行
Polars LazyFrames 讓你可以建立一連串操作（篩選、分組、排序），而不必立即執行。 Polars 在執行前會優化整個鏈條，這比逐步應用更快。
def query_to_lazy(cursor, query: str, params: dict = None) -> pl.LazyFrame:
"""Execute query and return a Polars LazyFrame."""
cursor.execute(query, params or {})
arrow_table = cursor.arrow()
return pl.from_arrow(arrow_table).lazy()
# Build a query plan without executing immediately
lf = query_to_lazy(cursor, "SELECT SalesOrderID, CustomerID, TotalDue, OrderDate FROM Sales.SalesOrderHeader")
result = (
lf.filter(pl.col("TotalDue") > 100)
.group_by("CustomerID")
.agg([
pl.col("TotalDue").sum().alias("TotalSpent"),
pl.col("SalesOrderID").count().alias("OrderCount")
])
.sort("TotalSpent", descending=True)
.collect() # Execute the optimized plan
)
print(result)
將 Polars DataFrames 寫入 Microsoft SQL
為防止 SQL 注入，請將識別碼加上引號
在 SQL 中，資料表和欄位名稱無法作為查詢參數傳遞。 當你建立帶有動態識別碼的 SQL 語句時，請用方括號包覆每個名稱，並跳脫任何嵌入 ] 字元，以防止 SQL 注入。
def quote_id(identifier: str) -> str:
"""Quote a Microsoft SQL identifier to prevent SQL injection.
Wraps the name in square brackets and escapes any embedded ] characters.
Raises ValueError if the identifier is empty or contains null bytes.
"""
if not identifier or "\x00" in identifier:
raise ValueError(f"Invalid identifier: {identifier!r}")
escaped = identifier.replace("]", "]]")
return f"[{escaped}]"
本節中的輔助函式在產生的 SQL 中，對所有資料表名稱和欄位名稱都使用 quote_id()。
插入 DataFrame 資料列
逐列方法會使用 iter_rows(named=True) 逐列疊代 DataFrame，並對每一列執行一次 INSERT。 這種方法很直接，但在處理大量資料時會較慢，因為每一列都需要與伺服器進行一次往返通訊。
def polars_to_sql(cursor, conn, df: pl.DataFrame, table: str) -> int:
"""Write Polars DataFrame to Microsoft SQL table."""
columns = df.columns
placeholders = ", ".join([f"%({col})s" for col in columns])
col_list = ", ".join([quote_id(col) for col in columns])
query = f"INSERT INTO {quote_id(table)} ({col_list}) VALUES ({placeholders})"
rows_inserted = 0
for row in df.iter_rows(named=True):
params = {k: (None if v is None else v) for k, v in row.items()}
cursor.execute(query, params)
rows_inserted += 1
conn.commit()
return rows_inserted
# Usage
cursor.execute("CREATE TABLE #PolarsInsert (Name NVARCHAR(100), Price DECIMAL(10,2), CategoryID INT)")
df = pl.DataFrame({
"Name": ["Product A", "Product B"],
"Price": [29.99, 49.99],
"CategoryID": [1, 2]
})
rows = polars_to_sql(cursor, conn, df, "#PolarsInsert")
print(f"Inserted {rows} rows")
大量插入（建議用於大型資料框）
對於大型資料框架，使用驅動程式的方法bulkcopy()，透過 TDS（Tabular Data Stream）協定批量傳送資料列，這是 Microsoft SQL 使用的原生線路協定。 這種方法可減少往返次數，且比逐列插入更快。
def polars_to_sql_bulk(conn, df: pl.DataFrame, table: str) -> int:
"""Bulk insert Polars DataFrame using BCP for best performance."""
rows = [tuple(None if v is None else v for v in row) for row in df.iter_rows()]
cursor = conn.cursor()
result = cursor.bulkcopy(table, rows)
conn.commit()
return result["rows_copied"]
# Usage
cursor.execute("CREATE TABLE ##PolarsBulk (Name NVARCHAR(50), Price FLOAT, CategoryID INT)")
conn.commit()
df = pl.DataFrame({
"Name": ["Product A", "Product B", "Product C"],
"Price": [29.99, 49.99, 19.99],
"CategoryID": [1, 2, 1]
})
rows = polars_to_sql_bulk(conn, df, "##PolarsBulk")
print(f"Bulk inserted {rows} rows")
從 Arrow 內存大量插入資料
此cursor.bulkcopy_arrow()方法載入 Arrow 資料時，不會先將其轉換成 Python 元組。 Polars 以 Arrow 格式儲存資料，並透過 Arrow C 資料介面匯出，因此你可以直接傳遞 DataFrame：
def polars_to_sql_arrow(conn, df: pl.DataFrame, table: str) -> int:
"""Bulk insert a Polars DataFrame through the Arrow path."""
cursor = conn.cursor()
result = cursor.bulkcopy_arrow(table, df)
return result["rows_copied"]
# bulkcopy_arrow() opens its own connection, so commit the table creation first.
conn.autocommit = True
cursor = conn.cursor()
cursor.execute("CREATE TABLE ##PolarsArrow (Name NVARCHAR(50), Price FLOAT, CategoryID INT)")
df = pl.DataFrame({
"Name": ["Product A", "Product B", "Product C"],
"Price": [29.99, 49.99, 19.99],
"CategoryID": [1, 2, 1]
})
rows = polars_to_sql_arrow(conn, df, "##PolarsArrow")
print(f"Bulk inserted {rows} rows")
df.to_arrow() 也可行，但 Polars 在轉換過程中會將字串欄位 string_view 從 到 重新 large_string 編碼，這樣就能複製所有字串資料。
每個 Arrow 欄位類型必須與其目的 SQL 欄位類型相容。 對於 金錢、 十進位和 數字 欄位，請使用 decimal128 欄位而非浮點欄位。 欲了解更多資訊，請參閱 Apache Arrow 整合。
資料分析模式
以下範例展示了結合 Microsoft SQL 查詢與 Polars 轉換的常見分析任務。
彙總查詢
此範例依子類別分組產品，並以 SQL 計算數量與價格統計，然後將摘要載入 Polars 資料框中：
def get_sales_summary(cursor) -> pl.DataFrame:
"""Get sales summary by subcategory."""
cursor.execute("""
SELECT
sc.Name AS SubcategoryName,
COUNT(*) AS ProductCount,
AVG(p.ListPrice) AS AvgPrice,
MIN(p.ListPrice) AS MinPrice,
MAX(p.ListPrice) AS MaxPrice
FROM Production.Product p
JOIN Production.ProductSubcategory sc ON p.ProductSubcategoryID = sc.ProductSubcategoryID
GROUP BY sc.Name
ORDER BY ProductCount DESC
""")
return pl.from_arrow(cursor.arrow())
df = get_sales_summary(cursor)
print(df)
時間序列分析
從 Microsoft SQL 載入時間序列資料，並使用 Polars 運算式新增計算欄位，例如移動平均。
def get_daily_sales(cursor, start_date: str, end_date: str) -> pl.DataFrame:
"""Get daily sales and compute rolling statistics."""
cursor.execute("""
SELECT
CAST(OrderDate AS DATE) AS Date,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS Revenue
FROM Sales.SalesOrderHeader
WHERE OrderDate BETWEEN %(start)s AND %(end)s
GROUP BY CAST(OrderDate AS DATE)
ORDER BY Date
""", {"start": start_date, "end": end_date})
df = pl.from_arrow(cursor.arrow())
# Add rolling 7-day average
df = df.with_columns(
pl.col("Revenue").rolling_mean(window_size=7).alias("RollingAvg")
)
return df
sales_df = get_daily_sales(cursor, "2013-01-01", "2013-12-31")
print(sales_df)
將 SQL 資料與本地檔案連結
你可以透過在 Polars 中與本地 CSV 檔案結合 Microsoft SQL 資料來豐富它。 將每個來源載入 DataFrame，並於記憶體中聯結。
# Load SQL data via Arrow
cursor.execute("SELECT c.CustomerID, p.FirstName, p.LastName FROM Sales.Customer c JOIN Person.Person p ON c.PersonID = p.BusinessEntityID")
customers = pl.from_arrow(cursor.arrow())
# Load local CSV
orders = pl.read_csv("orders_export.csv")
# Join in Polars
result = customers.join(orders, on="CustomerID", how="inner")
print(result)
ETL 模式
結合 Microsoft SQL 查詢與 Polars 轉換，建立擷取、轉換與載入（ETL）管線。 Polars 運算式負責轉換步驟，而 bulkcopy() 則負責載入。
擷取、轉換、載入
此範例透過 Arrow 擷取活躍客戶資料，應用業務分區邏輯與 Polars 表達式，並以批量文案載入結果。
def etl_pipeline(source_cursor, dest_conn):
"""ETL pipeline using Polars transformations."""
# Extract: derive a per-customer summary from order history via Arrow
source_cursor.execute("""
SELECT
CustomerID,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS TotalSpent
FROM Sales.SalesOrderHeader
WHERE OrderDate > DATEADD(YEAR, -1, (SELECT MAX(OrderDate) FROM Sales.SalesOrderHeader))
GROUP BY CustomerID
""")
df = pl.from_arrow(source_cursor.arrow())
df = df.with_columns(pl.col("TotalSpent").cast(pl.Float64))
# Transform with Polars expressions
df = df.with_columns([
pl.when(pl.col("TotalSpent") > 1000).then(pl.lit("Platinum"))
.when(pl.col("TotalSpent") > 500).then(pl.lit("Gold"))
.when(pl.col("TotalSpent") > 100).then(pl.lit("Silver"))
.otherwise(pl.lit("Bronze"))
.alias("CustomerSegment"),
(pl.col("TotalSpent") / pl.col("OrderCount").clip(lower_bound=1))
.alias("AvgOrderValue"),
(pl.col("TotalSpent") > 500).alias("IsHighValue")
])
# Load via bulk copy into the destination table
dest_cursor = dest_conn.cursor()
dest_cursor.execute("""
CREATE TABLE ##CustomerAnalytics (
CustomerID INT,
CustomerSegment NVARCHAR(20),
AvgOrderValue FLOAT,
IsHighValue BIT
)
""")
dest_conn.commit()
load_df = df.select(["CustomerID", "CustomerSegment", "AvgOrderValue", "IsHighValue"])
polars_to_sql_bulk(dest_conn, load_df, "##CustomerAnalytics")
return len(df)
效能祕訣
以下建議能幫助你充分利用 mssql-python 和 Polars 的組合。
讓 Microsoft SQL 來處理繁重的工作
Microsoft SQL 在彙總、過濾和聯結方面，比起透過網路傳輸所有原始資料並在本機端以 Python 處理，速度更快。 盡可能讓 Microsoft SQL 處理繁重工作，只透過網路傳輸所需的資料，並使用 Polars 進行那些在 Python 中處理起來更方便的分析與轉換。
# Avoid: pulling all rows over the wire to aggregate locally in Polars
df = query_to_polars_arrow(cursor, "SELECT * FROM Production.Product WHERE Color IS NOT NULL") # transfers entire table
summary = df.group_by("Color").agg(pl.col("ListPrice").sum()) # aggregation that SQL can do faster
# Better: push the aggregation into SQL and transfer only the summary
df = query_to_polars_arrow(cursor, """
SELECT Color AS Category, SUM(ListPrice) AS TotalAmount
FROM Production.Product
WHERE Color IS NOT NULL
GROUP BY Color
""")
所有讀取操作都使用 Arrow
基於箭頭的傳輸避免建立中間的 Python 物件，減少記憶體使用並提升吞吐量。 對於超過幾列的結果集，建議 cursor.arrow() 使用比手動逐列轉換更為可行的做法。
# Suboptimal: Row-by-row conversion
cursor.execute("SELECT * FROM Production.TransactionHistory")
columns = [col[0] for col in cursor.description]
rows = cursor.fetchall()
df = pl.DataFrame({col: [row[i] for row in rows] for i, col in enumerate(columns)})
# Better: Arrow-based transfer
cursor.execute("SELECT * FROM Production.TransactionHistory")
df = pl.from_arrow(cursor.arrow())