Schnellstart: Massenkopie von Daten zwischen SQL-Datenbanken in Fabric mit dem Mssql-Python-Treiber

In dieser Schnellstartanleitung verwenden Sie den mssql-python Treiber zum Massenkopien von Daten zwischen SQL-Datenbanken in Fabric. Die Anwendung lädt Tabellen aus einem Quelldatenbankschema in lokale Parkettdateien mithilfe von Apache Arrow herunter und lädt sie dann mithilfe der hochleistungsorientierten bulkcopy Methode in eine Zieldatenbank hoch. Sie können dieses Muster verwenden, um Daten zwischen SQL-Datenbanken in Fabric zu migrieren, zu replizieren oder zu transformieren.

Der mssql-python Treiber erfordert keine externen Abhängigkeiten von Windows-Computern. Der Treiber installiert alles, was er mit einer einzigen pip Installation benötigt, sodass Sie die neueste Version des Treibers für neue Skripts verwenden können, ohne andere Skripts zu unterbrechen, die Sie nicht zum Upgrade und Test benötigen.

mssql-python-Dokumentation | mssql-python-Quellcode | Paket (PyPI) | UV

Voraussetzungen

Laden Sie AdventureWorks-Beispieldaten in Ihre SQL-Datenbank als Quelldatenbank.
(Optional) Eine zweite SQL-Datenbank in Fabric, die als Ziel verwendet werden soll. Der Benutzer muss über die Berechtigung zum Erstellen und Schreiben in Tabellen verfügen. Wenn Sie nicht über eine zweite Datenbank verfügen, können Sie die Zielverbindungszeichenfolge so ändern, dass sie auf dieselbe Datenbank verweist und ein anderes Schema für die Zieltabellen verwendet.
Python 3
- Wenn Sie noch nicht über Python verfügen, installieren Sie den Python-Runtime - und Pip-Paket-Manager aus python.org.
- Möchten Sie Ihre eigene Umgebung nicht verwenden? Öffnen Sie als Devcontainer mit GitHub Codespaces.
Visual Studio Code mit den folgenden Erweiterungen:
- Python erweiterung für Visual Studio Code
Azure Command-Line Interface (CLI) – Erforderlich für die kennwortlose Authentifizierung unter macOS und Linux.
Wenn Sie uv noch nicht haben, installieren Sie uv gemäß den Anweisungen von https://docs.astral.sh/uv/getting-started/installation/.

Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen.

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

Erstellen Sie das Projekt, und führen Sie den Code aus.

Ein neues Projekt erstellen
Hinzufügen von Abhängigkeiten
Starten von Visual Studio Code
Pyproject.toml aktualisieren
Aktualisieren von main.py
Speichern der Verbindungszeichenfolgen
Verwenden Sie 'uv run', um das Skript auszuführen

Neues Projekt erstellen

Öffnen Sie eine Eingabeaufforderung in Ihrem Entwicklungsverzeichnis. Wenn Sie kein Verzeichnis haben, erstellen Sie ein neues Verzeichnis namens python, scriptsusw. Vermeiden Sie Ordner auf Ihrem OneDrive, die Synchronisierung kann die Verwaltung Ihrer virtuellen Umgebung beeinträchtigen.

Erstellen Sie ein neues Projekt mit uv.

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

Hinzufügen von Abhängigkeiten

Im selben Verzeichnis installieren Sie die Pakete mssql-python, python-dotenv und pyarrow.

uv add mssql-python python-dotenv pyarrow

Starten Sie Visual Studio Code.

Führen Sie im selben Verzeichnis den folgenden Befehl aus.

code .

Pyproject.toml aktualisieren

Das pyproject.toml enthält die Metadaten für Ihr Projekt. Öffnen Sie die Datei in Ihrem bevorzugten Editor.
Überprüfen Sie den Inhalt der Datei. Es sollte mit diesem Beispiel vergleichbar sein. Beachten Sie die Python-Version und Abhängigkeit, denn mssql-python verwendet >=, um eine Mindestversion zu definieren. Wenn Sie eine genaue Version bevorzugen, ändern Sie die >= Vorversionsnummer in ==. Die aufgelösten Versionen der einzelnen Pakete werden dann im uv.lock gespeichert. Die Sperrdatei stellt sicher, dass Entwickler, die am Projekt arbeiten, konsistente Paketversionen verwenden. Außerdem wird sichergestellt, dass derselbe Satz von Paketversionen beim Verteilen des Pakets an Endbenutzer verwendet wird. Sie sollten die uv.lock Datei nicht bearbeiten.
```
[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",
]
```

Aktualisieren Sie die Beschreibung so, dass sie aussagekräftiger ist.

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

Speichern und schließen Sie die Datei.

Aktualisieren von main.py

Öffnen Sie die Datei mit dem Namen main.py. Es sollte mit diesem Beispiel vergleichbar sein.

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

if __name__ == "__main__":
    main()

Ersetzen Sie den Inhalt der Datei durch die folgenden Codeblöcke. Jeder Block baut auf dem vorherigen auf und sollte der Reihenfolge nach zu main.py hinzugefügt werden.

Tipp

Wenn Visual Studio Code Probleme beim Beheben von Paketen hat, müssen Sie den Interpreter aktualisieren, um die virtuelle Umgebung zu verwenden.

Fügen Sie am Anfang von main.py die Importe und Konstanten hinzu. Das Skript verwendet mssql_python für die Datenbankkonnektivität, pyarrow und pyarrow.parquet für die Verarbeitung von Spaltendaten und die Parquet-Datei-E/A, python-dotenv zum Laden von Verbindungszeichenfolgen aus einer .env-Datei, und ein kompiliertes Regex-Muster, das SQL-Bezeichner überprüft, um Injektionen zu verhindern.

"""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

Fügen Sie unterhalb der Importe die SQL-zu-Arrow-Typzuordnung hinzu. Dieses Wörterbuch übersetzt SQL Server-Spaltentypen in die entsprechenden Apache Arrow-Typen, sodass die Datentreue beim Schreiben in Parquet beibehalten wird. Die beiden Hilfsfunktionen erstellen präzise SQL-Typ-Zeichenfolgen (z. B. NVARCHAR(100) oder DECIMAL(18,2)) aus INFORMATION_SCHEMA Metadaten und ermitteln den passenden Arrow-Typ für jede Spalte.

_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

Fügen Sie die Schemaintrospection- und DDL-Generierungsfunktionen hinzu. _get_arrow_schema fragt INFORMATION_SCHEMA.COLUMNS mithilfe von parameterisierten Abfragen ab, erstellt ein Pfeilschema und speichert den ursprünglichen SQL-Typ als Feldmetadaten, damit die Zieltabelle mit exakten Spaltendefinitionen neu erstellt werden kann. _create_table_ddl liest diese Metadaten zurück, um DDL zu generieren DROP/CREATE TABLE . Der Typ timestamp (Rowversion) wird zu VARBINARY(8) neu zugeordnet, weil er automatisch generiert wird und nicht direkt eingefügt werden kann.

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);"
    )

Fügen Sie die Downloadfunktion hinzu. download_table liest Zeilen aus einer Quelltabelle in Batches mithilfe von fetchmany, konvertiert jeden Wert in einen Arrow-kompatiblen Python-Typ und schreibt Rekord-Batches inkrementell in eine Parquet-Datei mit pq.ParquetWriter. Bei diesem Ansatz wird verhindert, dass die gesamte Tabelle in den Arbeitsspeicher geladen wird. Die Funktion verwendet zwei separate Cursor: einen zum Lesen von Spaltenmetadaten und einen zum Streamen der Daten. Wenn die Tabelle leer ist, wird sie frühzeitig zurückgegeben.

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

Fügen Sie den Anreicherungshaken hinzu. enrich_parquet ist ein Platzhalter, in dem Sie Transformationen, abgeleitete Spalten oder Verknüpfungen zu Daten hinzufügen können, bevor sie hochgeladen werden. In dieser Schnellstartanleitung handelt es sich um eine no-op, die den Dateipfad unverändert zurückgibt.
```
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
```

Fügen Sie die Upload-Funktion hinzu. upload_parquet liest das Arrow-Schema aus der Parquet-Datei, generiert und führt DROP/CREATE TABLE DDL aus, um das Ziel vorzubereiten, liest dann die Datei in Batches und ruft cursor.bulkcopy() für hochleistungsfähige Masseneinfügungen auf. Die table_lock=True-Option verbessert den Durchsatz, indem die Sperrkonkurrenz minimiert wird. Nach Abschluss des Uploads wird eine SELECT COUNT(*) Funktion ausgeführt, um die Übereinstimmungen der Zeilenanzahl zu überprüfen.

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

Fügen Sie die Orchestrierungsfunktion hinzu. transfer_tables verbindet die drei Phasen zusammen. Sie stellt eine Verbindung mit der Quelldatenbank her, ermittelt alle Basistabellen im angegebenen Schema über INFORMATION_SCHEMA.TABLES, lädt jede in eine lokale Parquet-Datei herunter, führt den Enrichment-Hook aus, stellt dann eine Verbindung mit der Zieldatenbank her und lädt jede Datei hoch.

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)

Fügen Sie schließlich den main Einstiegspunkt hinzu. Sie lädt die .env Datei, ruft transfer_tables die Quell- und Zielverbindungszeichenfolgen auf und druckt die gesamte verstrichene Zeit.

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()

Speichern Sie und schließen Sie main.py.

Speichern der Verbindungszeichenfolgen

Öffnen Sie die .gitignore Datei, und fügen Sie einen Ausschluss für Dateien hinzu .env . Ihre Datei sollte mit diesem Beispiel vergleichbar sein. Achten Sie darauf, sie zu speichern und zu schließen, wenn Sie fertig sind.
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env
```
Erstellen Sie im aktuellen Verzeichnis eine neue Datei mit dem Namen .env.
Fügen Sie in der .env Datei Einträge für die Quell- und Zielverbindungsstrings hinzu. Ersetzen Sie die Platzhalterwerte durch die tatsächlichen Server- und Datenbanknamen.
```
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"
```
Tipp

Verwenden Sie für die SQL-Datenbank in Fabric die ODBC-Verbindungszeichenfolge von der Registerkarte "Verbindungszeichenfolgen" ohne Treiberinformationen. Weitere Informationen finden Sie unter Herstellen einer Verbindung mit Ihrer SQL-Datenbank in Fabric.

Tipp

Unter macOS funktionieren sowohl ActiveDirectoryInteractive als auch ActiveDirectoryDefault für die Microsoft Entra-Authentifizierung. ActiveDirectoryInteractive fordert Sie auf, sich jedes Mal anzumelden, wenn Sie das Skript ausführen. Um wiederholte Anmeldeaufforderungen zu vermeiden, melden Sie sich einmal über die Azure CLI an, indem Sie sie ausführen az login, und verwenden Sie ActiveDirectoryDefaultdann die zwischengespeicherten Anmeldeinformationen.

Verwenden Sie "uv run", um das Skript auszuführen.

Führen Sie im Terminalfenster vor oder in einem neuen Terminalfenster, das im selben Verzeichnis geöffnet ist, den folgenden Befehl aus.

 uv run main.py

Dies ist die erwartete Ausgabe, wenn das Skript abgeschlossen ist.

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

Stellen Sie eine Verbindung mit der Zieldatenbank her, und stellen Sie sicher, dass die Tabellen und Daten erfolgreich erstellt wurden. Weitere Optionen zum Herstellen einer Verbindung finden Sie unter Herstellen einer Verbindung mit Ihrer SQL-Datenbank in Fabric.
Um Ihr Skript auf einem anderen Computer bereitzustellen, kopieren Sie alle Dateien mit Ausnahme des .venv Ordners auf den anderen Computer. Die virtuelle Umgebung wird mit der ersten Ausführung neu erstellt.

Funktionsweise des Codes

Die Anwendung führt eine vollständige Roundtrip-Datenübertragung in drei Phasen durch:

Download: Stellt eine Verbindung mit der Quelldatenbank her, liest Spaltenmetadaten aus INFORMATION_SCHEMA.COLUMNS, erstellt ein Apache Arrow-Schema und lädt dann jede Tabelle in Batches in eine lokale Parkettdatei herunter.pq.ParquetWriter
Anreicher (optional): Stellt einen Hook (enrich_parquet) bereit, an dem Sie Transformationen, abgeleitete Spalten oder Verknüpfungen vor dem Hochladen hinzufügen können.
Upload: Liest jede Parquet-Datei in Batches, erstellt die Tabelle in der Zieldatenbank mithilfe von DDL, die aus Arrow-Schemametadaten generiert wurde, und verwendet dann cursor.bulkcopy() für leistungsstarke Masseneinfügungen.

Nächster Schritt

Besuchen Sie das mssql-python GitHub-Treiberrepository für weitere Beispiele, um Ideen beizutragen oder Probleme zu melden.

mssql-python-Treiber auf GitHub

Feedback

War diese Seite hilfreich?

Last updated on 2026-03-18