快速入門：使用 mssql-python 驅動程式在 Fabric 中的 SQL 資料庫間批量複製資料

在這個快速入門中，你會使用 mssql-python 驅動程式在 Fabric 的 SQL 資料庫間批量複製資料。應用程式會使用 Apache Arrow 從來源資料庫結構下載資料表到本地 Parquet 檔案，然後使用高效能 bulkcopy 方法上傳至目標資料庫。你可以利用此模式在 Fabric 中的 SQL 資料庫間遷移、複製或轉換資料。

在 Windows 電腦上，mssql-python 驅動程式不需要任何外部相依性。驅動程式會透過單一 pip 安裝來安裝所需的所有內容，讓您可以將最新版本的驅動程式用於新腳本，而不會中斷您沒有時間升級和測試的其他腳本。

MSSQL-Python 文件 | MSSQL-Python 原始碼 | 套件（PyPI） | 紫外線

先決條件

將 AdventureWorks 範例資料載入你的 SQL 資料庫，作為來源資料庫。
（可選）在 Fabric 裡再用第二個 SQL 資料庫作為目的地。使用者必須擁有建立及寫入資料表的權限。如果你沒有第二個資料庫，可以把目的的連線字串改成指向同一個資料庫，並用不同的 schema 來管理目的資料表。
蟒蛇3
- 如果你還沒有 Python，建議安裝 Python 執行環境 和 Pip 套件管理器 ，從 python.org 安裝。
- 不想使用自己的系統環境？使用 GitHub Codespaces 開啟為開發容器。
Visual Studio Code 使用下列擴充套件：
- Visual Studio Code 的 Python 擴充功能
Azure Command-Line 介面（CLI） -macOS 和 Linux 無密碼認證所必需。
如果您還沒有uv，請按照uv中的指示安裝https://docs.astral.sh/uv/getting-started/installation/。

安裝一次性作業系統特定先決條件。

apk add libtool krb5-libs krb5-dev

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

dnf install -y libtool-ltdl krb5-libs

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

zypper install -y libltdl7

brew install openssl

建立新專案

在開發目錄中開啟命令提示字元。如果您沒有，請建立一個名為、 python等的新scripts目錄。避免在 OneDrive 上放置資料夾，同步可能會幹擾虛擬環境的管理。

建立新的專案使用uv。

uv init mssql-python-bcp-qs
cd mssql-python-bcp-qs

新增依賴性

在相同的目錄中，安裝 mssql-python、 python-dotenv和 pyarrow 套件。

uv add mssql-python python-dotenv pyarrow

啟動 Visual Studio Code

在相同的目錄中，執行下列命令。

code .

更新 pyproject.toml

pyproject.toml 包含專案的中繼資料。在您最喜歡的編輯器中打開該文件。
檢閱檔案的內容。它應該類似於這個例子。請注意 Python 的版本和相依性，mssql-python 使用 >= 來定義最低版本。如果您偏好確切的版本，請將版本號碼之前的變更 >= 為 ==。然後，每個套件的解析版本會儲存在 uv.lock 中。鎖定檔案可確保處理專案的開發人員使用一致的套件版本。它也會確保在將套件分發給一般使用者時，使用完全相同的套件版本集。您不應該編輯檔案 uv.lock 。
```
[project]
name = "mssql-python-bcp-qs"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "mssql-python>=1.4.0",
    "python-dotenv>=1.1.1",
    "pyarrow>=19.0.0",
]
```

更新描述以更具描述性。

description = "Bulk copies data between SQL databases using mssql-python and Apache Arrow"

儲存並關閉檔案。

更新 main.py

開啟名為 main.py的檔案。它應該類似於這個例子。

def main():
    print("Hello from mssql-python-bcp-qs!")

if __name__ == "__main__":
    main()

將檔案內容替換為以下程式碼區塊。每個區塊都是建構在前一個區塊上，應該依序添加至 main.py。

小提示

如果 Visual Studio Code 在解析套件時遇到問題，您必須更新解譯器才能使用虛擬環境。

在 main.py 的頂部加入導入和常數。該腳本用於mssql_python資料庫連接、pyarrowpyarrow.parquet欄位資料處理及 Parquet 檔案 I/O、python-dotenv從檔案載入連接字串.env，以及已編譯的正則表達式，以驗證 SQL 識別碼以防止注入。

"""Round-trip: download tables from a source DB/schema to parquet, upload to a destination DB/schema."""

import os
import re
import time
from uuid import UUID

import pyarrow as pa
import pyarrow.parquet as pq
from dotenv import load_dotenv
import mssql_python

BATCH_SIZE = 64_000
_SAFE_IDENT = re.compile(r"^[A-Za-z0-9_]+$")


def _validate_ident(name: str) -> str:
    if not _SAFE_IDENT.match(name):
        raise ValueError(f"Unsafe SQL identifier: {name!r}")
    return name

在匯入的下方，新增從 SQL 到 Arrow 類型的映射。此字典會將 SQL Server 欄位類型轉換為其 Apache Arrow 對應的欄位，以確保在寫入 Parquet 時保持資料的完整性。這兩個輔助函式會從NVARCHAR(100)元資料建立精確的 SQL 類型字串（例如 DECIMAL(18,2) 或 INFORMATION_SCHEMA），並為每個欄位解析相符的 Arrow 類型。

_SQL_TO_ARROW = {
    "bit": pa.bool_(),
    "tinyint": pa.uint8(),
    "smallint": pa.int16(),
    "int": pa.int32(),
    "bigint": pa.int64(),
    "float": pa.float64(),
    "real": pa.float32(),
    "smallmoney": pa.decimal128(10, 4),
    "money": pa.decimal128(19, 4),
    "date": pa.date32(),
    "datetime": pa.timestamp("ms"),
    "datetime2": pa.timestamp("us"),
    "smalldatetime": pa.timestamp("s"),
    "uniqueidentifier": pa.string(),
    "xml": pa.string(),
    "image": pa.binary(),
    "binary": pa.binary(),
    "varbinary": pa.binary(),
    "timestamp": pa.binary(),
}


def _sql_type_str(data_type: str, max_length: int, precision: int, scale: int) -> str:
    """Build the exact SQL type string from INFORMATION_SCHEMA metadata."""
    dt = data_type.lower()
    if dt in ("char", "varchar", "nchar", "nvarchar", "binary", "varbinary"):
        length = "MAX" if max_length == -1 else str(max_length)
        return f"{dt.upper()}({length})"
    if dt in ("decimal", "numeric"):
        return f"{dt.upper()}({precision},{scale})"
    return dt.upper()


def _arrow_type(sql_type: str, precision: int, scale: int) -> pa.DataType:
    sql_type = sql_type.lower()
    if sql_type in _SQL_TO_ARROW:
        return _SQL_TO_ARROW[sql_type]
    if sql_type in ("decimal", "numeric"):
        return pa.decimal128(precision, scale)
    if sql_type in ("char", "varchar", "nchar", "nvarchar", "text", "ntext", "sysname"):
        return pa.string()
    return pa.string()


def _convert_value(v):
    """Convert a SQL value to an Arrow-compatible Python type."""
    if isinstance(v, UUID):
        return str(v)
    return v

加入結構內省和 DDL 生成函數。 _get_arrow_schema 使用參數化查詢對 INFORMATION_SCHEMA.COLUMNS 進行查詢，建立 Arrow 架構，並將原始 SQL 型別儲存為欄位元資料，以便目的資料表能以精確的欄位定義重建。 _create_table_ddl 會讀取該中繼資料以產生 DROP/CREATE TABLE DDL。 timestamp （rowversion）類型會被重新映射到VARBINARY(8)，因為它是自動產生的，無法直接插入。

def _get_arrow_schema(cursor, schema_name: str, table_name: str) -> pa.Schema:
    """Build an Arrow schema from INFORMATION_SCHEMA.COLUMNS.

    Stores the original SQL type as field metadata so the round-trip
    CREATE TABLE can reproduce exact column definitions.
    """
    cursor.execute(
        "SELECT COLUMN_NAME, DATA_TYPE, "
        "COALESCE(CHARACTER_MAXIMUM_LENGTH, 0), "
        "COALESCE(NUMERIC_PRECISION, 0), "
        "COALESCE(NUMERIC_SCALE, 0), "
        "IS_NULLABLE "
        "FROM INFORMATION_SCHEMA.COLUMNS "
        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? "
        "ORDER BY ORDINAL_POSITION",
        (schema_name, table_name),
    )
    rows = cursor.fetchall()
    if not rows:
        raise ValueError(f"No columns found for {schema_name}.{table_name}")
    fields = []
    for col_name, data_type, max_len, precision, scale, nullable in rows:
        arrow_t = _arrow_type(data_type, precision, scale)
        sql_t = _sql_type_str(data_type, max_len, precision, scale)
        fields.append(
            pa.field(
                col_name, arrow_t,
                nullable=(nullable == "YES"),
                metadata={"sql_type": sql_t},
            )
        )
    return pa.schema(fields)


def _create_table_ddl(target: str, schema: pa.Schema) -> str:
    """Build DROP/CREATE TABLE DDL from Arrow schema with SQL type metadata."""
    col_defs = []
    for f in schema:
        sql_t = f.metadata[b"sql_type"].decode()
        # timestamp/rowversion is auto-generated and not insertable
        if sql_t == "TIMESTAMP":
            sql_t = "VARBINARY(8)"
        null = "" if f.nullable else " NOT NULL"
        col_defs.append(f"[{f.name}] {sql_t}{null}")
    col_defs_str = ",\n    ".join(col_defs)
    return (
        f"IF OBJECT_ID('{target}', 'U') IS NOT NULL DROP TABLE {target};\n"
        f"CREATE TABLE {target} (\n    {col_defs_str}\n);"
    )

新增下載功能。 download_table從來源資料表中以批次方式讀取資料列，將fetchmany每個值轉換為與 Arrow 相容的 Python 型別，並以增量方式將記錄批次寫入 Parquet 檔案。pq.ParquetWriter 此方法避免將整個資料表載入記憶體。此函式使用兩個獨立游標：一個用於讀取欄位元資料，另一個用於串流資料。如果表格是空的，則會提前返回。

def download_table(conn, schema_name: str, table_name: str, parquet_file: str) -> int:
    """Download a SQL table to a parquet file. Returns row count (0 if empty)."""
    _validate_ident(table_name)
    source = f"{schema_name}.[{table_name}]"

    with conn.cursor() as cursor:
        schema = _get_arrow_schema(cursor, schema_name, table_name)

    n_cols = len(schema)
    row_count = 0
    t0 = time.perf_counter()

    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {source}")
        with pq.ParquetWriter(parquet_file, schema) as writer:
            while True:
                rows = cursor.fetchmany(BATCH_SIZE)
                if not rows:
                    break
                columns = [[] for _ in range(n_cols)]
                for row in rows:
                    for i in range(n_cols):
                        columns[i].append(_convert_value(row[i]))
                arrays = [
                    pa.array(columns[i], type=schema.field(i).type)
                    for i in range(n_cols)
                ]
                batch = pa.record_batch(arrays, schema=schema)
                writer.write_batch(batch)
                row_count += len(rows)

    if row_count == 0:
        os.remove(parquet_file)
        return 0

    elapsed = time.perf_counter() - t0
    rate = f"{int(row_count / elapsed):,} rows/sec" if elapsed > 0 else "n/a"
    print(
        f"{schema_name}.{table_name} → {parquet_file}: {row_count:,} rows downloaded "
        f"in {elapsed:.2f}s ({rate})"
    )
    return row_count

添加擴充掛勾。 enrich_parquet 是一個佔位符，你可以在資料上傳前加入轉換、派生欄位或連接。在這個快速入門中，它是不執行任何操作，會回傳不變的檔案路徑。

def enrich_parquet(parquet_file: str) -> str:
    """Enrich a parquet file before upload. Returns the (possibly new) file path."""
    # TODO: add transformations, derived columns, joins, etc.
    print(f"Enriching {parquet_file} (no-op)")
    return parquet_file

新增上傳功能。 upload_parquet 從 Parquet 檔案讀取 Arrow 資料架構，產生並執行 DROP/CREATE TABLE DDL 以準備目的端，然後批次讀取檔案並呼叫 cursor.bulkcopy() 進行高效能的批量插入。此 table_lock=True 選項透過最小化鎖爭用以提升吞吐量。上傳完成後，函式會執行 a SELECT COUNT(*) 以驗證列數是否相符。

def upload_parquet(conn, parquet_file: str, target: str) -> int:
    """Upload a parquet file into a SQL table via BCP. Returns row count."""
    # ── Create target table from parquet schema ──
    pf_schema = pq.read_schema(parquet_file)
    with conn.cursor() as cursor:
        cursor.execute(_create_table_ddl(target, pf_schema))
    conn.commit()

    # ── Bulk insert ──
    uploaded = 0
    t0 = time.perf_counter()
    with pq.ParquetFile(parquet_file) as pf:
        with conn.cursor() as cursor:
            for batch in pf.iter_batches(batch_size=BATCH_SIZE):
                rows = zip(*(col.to_pylist() for col in batch.columns))
                cursor.bulkcopy(
                    target, rows, batch_size=BATCH_SIZE,
                    table_lock=True, timeout=3600,
                )
                uploaded += batch.num_rows
    conn.commit()
    elapsed = time.perf_counter() - t0

    # ── Verify ──
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT COUNT(*) FROM {target}")
        count = cursor.fetchone()[0]

    rate = f"{int(uploaded / elapsed):,} rows/sec" if elapsed > 0 else "n/a"
    print(
        f"{parquet_file} → {target}: {uploaded:,} rows uploaded "
        f"in {elapsed:.2f}s ({rate}) "
        f"| verified: {count:,}"
    )
    return uploaded

加入編排功能。 transfer_tables 將三個階段串連起來。它連接到來源資料庫，透過 INFORMATION_SCHEMA.TABLES 發現該結構中的所有基表，將每個表格下載為本地的 Parquet 檔案，然後執行資料豐富處理，接著連接到目標資料庫並上傳所有檔案。

def transfer_tables(
    source_conn_str: str,
    dest_conn_str: str,
    source_schema: str,
    dest_schema: str,
) -> None:
    """Download all tables from source DB/schema to parquet, upload to dest DB/schema."""
    _validate_ident(source_schema)
    _validate_ident(dest_schema)

    parquet_dir = source_schema
    os.makedirs(parquet_dir, exist_ok=True)

    # ── Download from source ──
    with mssql_python.connect(source_conn_str) as src_conn:
        with src_conn.cursor() as cursor:
            cursor.execute(
                "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES "
                "WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE' "
                "ORDER BY TABLE_NAME",
                (source_schema,),
            )
            tables = [row[0] for row in cursor.fetchall()]

        print(f"Found {len(tables)} {source_schema} tables: {', '.join(tables)}\n")

        parquet_files = []
        for table_name in tables:
            parquet_file = os.path.join(parquet_dir, f"{table_name}.parquet")
            row_count = download_table(src_conn, source_schema, table_name, parquet_file)
            if row_count == 0:
                print(f"{source_schema}.{table_name}: empty, skipping")
            else:
                parquet_files.append((table_name, parquet_file))

    # ── Enrich parquet files ──
    enriched_files = []
    for table_name, parquet_file in parquet_files:
        enriched_file = enrich_parquet(parquet_file)
        enriched_files.append((table_name, enriched_file))

    # ── Upload to destination ──
    with mssql_python.connect(dest_conn_str) as dest_conn:
        for table_name, parquet_file in enriched_files:
            target = f"{dest_schema}.[{table_name}]"
            upload_parquet(dest_conn, parquet_file, target)

最後，加入 main 入口點。它會載 .env 入檔案，呼叫 transfer_tables 來源與目的連接字串，並列印總經過時間。

def main():
    load_dotenv()
    t_start = time.perf_counter()

    transfer_tables(
        source_conn_str=os.environ["SOURCE_CONNECTION_STRING"],
        dest_conn_str=os.environ["DEST_CONNECTION_STRING"],
        source_schema="SalesLT",
        dest_schema="dbo",
    )

    print(f"Total: {time.perf_counter() - t_start:.2f}s")


if __name__ == "__main__":
    main()

儲存後關閉 main.py。

保存連接線

開啟 .gitignore 檔案，並為 .env 檔案新增排除項目。您的檔案應該類似於此範例。請務必儲存並在完成後將其關閉。

# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

在目前目錄中，建立名為 .env的新檔案。
在 .env 檔案中，新增您來源與目的地連接字串的項目。把佔位符值替換成你實際的伺服器和資料庫名稱。
```
SOURCE_CONNECTION_STRING="Server=<source_server_name>;Database=<source_database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
DEST_CONNECTION_STRING="Server=<dest_server_name>;Database=<dest_database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
```
小提示

針對 Fabric 中的 SQL 資料庫，請使用 [連接字串] 索引標籤中的 ODBC 連接字串，而不使用 DRIVER 資訊。欲了解更多資訊，請參閱在 Fabric 中連接至您的 SQL 資料庫。

小提示

在 macOS 上，兩者都ActiveDirectoryInteractiveActiveDirectoryDefault適用於 Microsoft Entra 認證。 ActiveDirectoryInteractive 每次執行腳本時都會提示你登入。避免反覆出現的登入提示，請先透過 Azure CLI 登入，然後使用 az login，即可重複使用快取憑證。

使用 uv run 執行腳本

在先前的終端機視窗中，或開啟至相同目錄的新終端機視窗中，執行下列命令。

 uv run main.py

以下是指令碼完成時的預期輸出。

Found 12 SalesLT tables: Address, Customer, CustomerAddress, ...

SalesLT.Address → SalesLT/Address.parquet: 450 rows downloaded in 0.15s (3,000 rows/sec)
...
SalesLT/Address.parquet → dbo.[Address]: 450 rows uploaded in 0.10s (4,500 rows/sec) | verified: 450
...
Total: 2.35s

連接目標資料庫並確認資料表和資料是否成功建立。想了解更多連線選項，請參見 Connect to your SQL database in Fabric。
若要將指令碼部署到另一台電腦，請將資料夾以外的 .venv 所有檔案複製到另一部電腦。虛擬環境會在第一次執行時重新建立。

程式碼運作方式

該應用程式可分三個階段進行完整的往返資料傳輸：

下載：連接到來源資料庫，從 INFORMATION_SCHEMA.COLUMNS 讀取欄位元資料，創建 Apache Arrow 架構，然後使用 pq.ParquetWriter 將每個資料表分批下載至本地的 Parquet 檔案。
Enrich （可選）：提供一個 hook （enrich_parquet），可在上傳前新增轉換、衍生欄位或連接。
上傳：批次讀取每個 Parquet 檔案，利用 Arrow 結構元資料產生的 DDL 在目標資料庫中重新建資料表，然後使用 cursor.bulkcopy() 進行高效能的批量插入。

下一個步驟

請造訪 mssql-python 驅動程式 GitHub 存放庫以取得更多範例，以提供想法或報告問題。

GitHub 上的 mssql-python 驅動程式

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-03-18