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 {})
reader = cursor.arrow_reader(batch_size=batch_size)
results = []
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")
資料分析模式
以下範例展示了結合 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())