在此快速入門中，使用mssql-python驅動程式內建的 Arrow 擷取方法，以欄位式 Apache Arrow 資料表形式取得 SQL Server 資料。 Arrow 的列式記憶體格式可實現高效能分析、與 pandas、Polars 和 DuckDB 的零複製互通，以及無需逐列建立 Python 物件的高效率 Parquet 檔案 I/O。
在 Windows 電腦上，mssql-python 驅動程式不需要任何外部相依性。 驅動程式只需一次 pip 安裝就能安裝所有需要的東西，所以你可以用最新版本的驅動程式來執行新腳本，而不會破壞你沒時間升級和測試的其他腳本。
MSSQL-Python 文件 | MSSQL-Python 原始碼 | 套件（PyPI） | uv
先決條件
Python 3.10 或更新版本
如果你還沒有 Python，建議安裝 Python 執行環境 和 Pip 套件管理器 ，從 python.org 安裝。
不想使用自己的系統環境？ 遵循容器與本地開發，建立可重現的開發容器或 GitHub Codespaces 環境。
Visual Studio Code 使用下列擴充套件：
Azure Command-Line 介面（CLI）用於 macOS 和 Linux 的無密碼認證。
如果你還沒有安裝<該軟體>，請依照
安裝說明進行 操作。SQL Server、Azure SQL Database 或 Fabric 中的 SQL 資料庫上的資料庫，具有
AdventureWorks2025範例結構描述和有效的連接字串。
安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。 完整平台細節請參見 安裝 mssql-python。
建立 SQL 資料庫
在以下平台建立或連接 SQL 資料庫：
建立專案並執行程式碼
建立新專案
在開發目錄中開啟命令提示字元。 如果你沒有，請建立一個新的目錄，例如
python或scripts。 避免在 OneDrive 上放置資料夾，因為同步可能會干擾虛擬環境的管理。-
uv init arrow-qs cd arrow-qs
新增依賴性
在同一目錄中安裝 mssql-python、python-dotenv、pyarrow 和 rich 套件。
uv add mssql-python python-dotenv pyarrow rich
啟動 Visual Studio Code
在相同的目錄中，執行下列命令。
code .
更新 pyproject.toml
pyproject.toml 檔案包含你專案的元資料。 在您最喜歡的編輯器中打開該文件。
檢閱檔案的內容。 它應該類似於這個例子。 請注意針對
mssql-python的 Python 版本和相依性，並使用>=來定義最低版本。 如果您偏好確切的版本，請將版本號碼之前的 變更>=為==。 然後，每個套件的解析版本會儲存在 uv.lock 中。 鎖定檔案確保開發者使用一致的套件版本。 提交pyproject.toml和uv.lock兩者，並在 CI 中執行經組織核准的相依性掃描工具。 不要直接編輯uv.lock檔案。[project] name = "arrow-qs" version = "0.1.0" description = "Add your description here" readme = "README.md" requires-python = ">=3.11" dependencies = [ "mssql-python>=1.15.0", "pyarrow>=19.0.0", "python-dotenv>=1.1.1", "rich>=14.1.0", ]更新描述以更具描述性。
description = "Fetch SQL Server data as Apache Arrow tables using mssql-python"儲存並關閉檔案。
更新 main.py
開啟名為
main.py的檔案。 它應該類似於這個例子。def main(): print("Hello from arrow-qs!") if __name__ == "__main__": main()將整個
main.py內容替換成以下程式碼。"""Fetch SQL Server data as Apache Arrow tables using mssql-python.""" from os import getenv import pyarrow as pa import pyarrow.parquet as pq from dotenv import load_dotenv from mssql_python import connect, Connection from rich.console import Console from rich.table import Table console = Console() def get_connection() -> Connection: """Create a connection using the connection string from .env.""" load_dotenv() conn_str = getenv("SQL_CONNECTION_STRING") if not conn_str: raise ValueError("SQL_CONNECTION_STRING not set in .env file") return connect(conn_str) def fetch_arrow_table(conn: Connection) -> pa.Table: """Run a query and return the full result as an Arrow Table.""" cursor = conn.cursor() cursor.execute(""" SELECT p.ProductID, p.Name, p.ProductNumber, p.Color, p.StandardCost, p.ListPrice, p.Size, p.Weight, p.SellStartDate, pc.Name AS Category FROM SalesLT.Product AS p INNER JOIN SalesLT.ProductCategory AS pc ON p.ProductCategoryID = pc.ProductCategoryID ORDER BY p.ListPrice DESC """) arrow_table = cursor.arrow() cursor.close() return arrow_table def fetch_arrow_batches(conn: Connection) -> pa.Table: """Stream results one batch at a time using arrow_batch().""" cursor = conn.cursor() cursor.execute(""" SELECT c.CustomerID, c.CompanyName, c.EmailAddress, COUNT(soh.SalesOrderID) AS OrderCount, SUM(soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalSpent FROM SalesLT.Customer AS c LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh ON c.CustomerID = soh.CustomerID GROUP BY c.CustomerID, c.CompanyName, c.EmailAddress ORDER BY TotalSpent DESC """) batches = [] while True: batch = cursor.arrow_batch() if batch is None or batch.num_rows == 0: break batches.append(batch) cursor.close() if not batches: return pa.table({}) return pa.Table.from_batches(batches) def fetch_with_reader(conn: Connection) -> pa.Table: """Stream results with arrow_reader() and combine the batches.""" cursor = conn.cursor() cursor.execute(""" SELECT soh.SalesOrderID, soh.OrderDate, (soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalDue, c.CompanyName FROM SalesLT.SalesOrderHeader AS soh INNER JOIN SalesLT.Customer AS c ON soh.CustomerID = c.CustomerID ORDER BY soh.OrderDate DESC """) reader = cursor.arrow_reader() batches = list(reader) cursor.close() if not batches: return pa.table({}) return pa.Table.from_batches(batches) def display_arrow_table(arrow_table: pa.Table, title: str, max_rows: int = 10) -> None: """Display an Arrow table using rich formatting.""" rich_table = Table(title=title) for name in arrow_table.column_names: rich_table.add_column(name, style="bright_white") for i in range(min(max_rows, arrow_table.num_rows)): row = [str(arrow_table.column(col)[i].as_py()) for col in range(arrow_table.num_columns)] rich_table.add_row(*row) if arrow_table.num_rows > max_rows: rich_table.add_row(*[f"... ({arrow_table.num_rows - max_rows} more rows)" if col == 0 else "" for col in range(arrow_table.num_columns)]) console.print(rich_table) console.print(f"\n[dim]Schema: {arrow_table.num_columns} columns, {arrow_table.num_rows} rows[/dim]\n") def save_to_parquet(arrow_table: pa.Table, file_path: str) -> None: """Save an Arrow table to a Parquet file.""" pq.write_table(arrow_table, file_path) console.print(f"[green]Saved {arrow_table.num_rows} rows to {file_path}[/green]\n") def main() -> None: conn = get_connection() # 1. Fetch entire result as an Arrow Table with cursor.arrow() console.rule("[bold]cursor.arrow() - Full table fetch[/bold]") products = fetch_arrow_table(conn) display_arrow_table(products, "Products (Top 10 by List Price)") # 2. Stream results in batches with cursor.arrow_batch() console.rule("[bold]cursor.arrow_batch() - Batch streaming[/bold]") customers = fetch_arrow_batches(conn) display_arrow_table(customers, "Customers by Total Spent") # 3. Stream results with cursor.arrow_reader() console.rule("[bold]cursor.arrow_reader() - Reader streaming[/bold]") orders = fetch_with_reader(conn) display_arrow_table(orders, "Recent Orders") # 4. Save to Parquet console.rule("[bold]Save to Parquet[/bold]") save_to_parquet(products, "products.parquet") # 5. Read back from Parquet and verify loaded = pq.read_table("products.parquet") console.print(f"[green]Read back {loaded.num_rows} rows from products.parquet[/green]") console.print(f"[dim]Schema: {loaded.schema}[/dim]\n") conn.close() if __name__ == "__main__": main()
儲存連接字串
開啟
.gitignore檔案，並為.env檔案新增排除項目。 您的檔案應該類似於此範例。 請務必儲存並在完成後將其關閉。# Python-generated files __pycache__/ *.py[oc] build/ dist/ wheels/ *.egg-info # Virtual environments .venv # Connection strings and secrets .env # Generated data files *.parquet在目前目錄中，建立名為
.env的新檔案。在
.env檔案中，新增一個名為SQL_CONNECTION_STRING的連接字串項目。 將此處的範例替換為您的實際連接字符串值。SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"Important
保持本地化
.env，且不使用原始碼控制。 對於 CI 和已部署的環境，請從平台的祕密儲存區中注入連線字串或其組成祕密，而不要在不同機器之間複製.env。Tip
你使用的 連接字串 很大程度取決於你連接的 SQL 資料庫類型。 如果您要連線到 Azure SQL 資料庫 或 Fabric 中的 SQL 資料庫，請使用 [連接字串] 索引標籤中的 ODBC 連接字串。您可能需要根據您的案例調整驗證類型。 如需連接字串及其語法的詳細資訊，請參閱 連接字串語法參考。
使用 uv run 執行腳本
Tip
在 macOS 上，兩者都ActiveDirectoryInteractiveActiveDirectoryDefault適用於 Microsoft Entra 認證。
ActiveDirectoryInteractive 每次執行腳本時都會提示你登入。 為避免重複登入提示，請透過 Azure CLI 執行 az login，然後使用 ActiveDirectoryDefault，重複使用快取的憑證。
在先前的終端機視窗中，或開啟至相同目錄的新終端機視窗中，執行下列命令。
uv run main.py該腳本示範了三種 Arrow 擷取方法：
cursor.arrow()回傳包含所有列的 completepyarrow.Table。 最適合用於中小型結果集，且需要將完整資料集載入記憶體的情況。cursor.arrow_batch()一次傳回一個pyarrow.RecordBatch。 最適合大型結果集，想要逐步處理資料而不把所有資料載入記憶體。cursor.arrow_reader()回傳一個串流讀取器，當你迭代時會產生pyarrow.RecordBatch物件。 將其用於管線式處理，或直接傳遞給可接受讀取器的函式庫。
腳本也會將產品資料儲存為 Parquet 檔案，並再讀取回來以驗證往返流程。
程式碼運作方式
連線：腳本從
.env檔案載入連線字串，並使用mssql_python.connect()建立連線。完整資料表擷取：
cursor.arrow()執行查詢並回傳整個結果集為pyarrow.Table。 驅動程式透過 Arrow C 資料介面在 C++ 層中轉換資料，繞過 Python 物件建立以提升效能。批次串流：
cursor.arrow_batch()每次通話回傳一個pyarrow.RecordBatch。 迴圈會收集批次直到沒有更多列，然後將它們合併成單一表格。 這種方法適用於大型資料集或想獨立處理每批資料時。串流讀取器：
cursor.arrow_reader()回傳一個讀取器，當你迭代時會產生pyarrow.RecordBatch物件。 與迴arrow_batch()圈相比，讀取器管理自己的迭代狀態，因此你不必自己測試耗盡的結果集。這個函式會讀取串流直到完成後關閉游標，因此不需要關閉讀取器。 提前停止閱讀時，請關閉閱讀器並繼續使用游標。 欲了解更多資訊，請參閱 Apache Arrow 整合。
Parquet I/O：
pyarrow.parquet.write_table()將 Arrow 表格儲存為壓縮後的 Parquet 檔案。 此格式保留欄位類型並支援高效的部分讀取。