Rychlý start: Hromadné kopírování pomocí ovladače mssql-python pro Python

V tomto rychlém průvodci použijete mssql-python ovladač k hromadnému kopírování dat mezi databázemi. Aplikace stáhne tabulky ze zdrojového schématu databáze do místních souborů Parquet pomocí Apache Arrow a pak je nahraje do cílové databáze pomocí vysoce výkonné bulkcopy metody. Tento vzor můžete použít k migraci, replikaci nebo transformaci dat mezi SQL Serverem, Azure SQL Database a SQL databází v prostředí Fabric.

Ovladač mssql-python nevyžaduje žádné externí závislosti na počítačích s Windows. Ovladač nainstaluje všechno, co potřebuje, s jednou pip instalací, což vám umožní používat nejnovější verzi ovladače pro nové skripty bez přerušení jiných skriptů, které nemáte čas upgradovat a testovat.

Dokumentace mssql-python, zdrojový kód mssql-python, Balíček (PyPI) uv

Předpoklady

Python 3.10 nebo novější
Pokud ještě nemáte Python, nainstalujte Python runtime a správce balíčků pip z python.org.
Nechcete používat vlastní prostředí? Postupujte podle Kontejner a místní vývoj a vytvořte si reprodukovatelný devcontainer nebo prostředí GitHub Codespaces.
Visual Studio Code s následujícími rozšířeními:
- Pythonové rozšíření pro Visual Studio Code
Azure Command-Line Interface (CLI) pro ověřování bez hesla v systému macOS a Linux.
Pokud ještě uvnemáte, postupujte podle pokynů k instalaci.
Zdrojová databáze s ukázkovým schématem AdventureWorksLT a platným připojovacím řetězcem.
Cílová databáze s platným připojovacím řetězcem. Uživatel musí mít oprávnění k vytváření a zápisu do tabulek. Pokud nemáte druhou databázi, můžete použít stejnou databázi a jiné schéma pro cílové tabulky.

Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace mssql-python.

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

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Vytvoření databáze SQL v řádu minut pomocí webu Azure Portal

Vytvoření nového projektu

Otevřete příkazový řádek ve vývojovém adresáři. Pokud žádný nemáte, vytvořte nový adresář, například python nebo scripts. Vyhněte se složkám na OneDrive, protože synchronizace může narušit správu virtuálního prostředí.

Vytvořte nový projekt pomocí uvpříkazu .

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

Přidejte závislosti

Ve stejném adresáři nainstalujte balíčky mssql-python, python-dotenv a pyarrow.

uv add mssql-python python-dotenv pyarrow

Spusťte Visual Studio Code

Ve stejném adresáři spusťte následující příkaz.

code .

Aktualizace pyproject.toml

Soubor pyproject.toml obsahuje metadata pro váš projekt. Otevřete soubor v oblíbeném editoru.
Zkontrolujte obsah souboru. Měl by se podobat tomuto příkladu. Všimněte si verze Pythonu a závislosti pro mssql-python, které používá >= k definování minimální verze. Pokud dáváte přednost přesné verzi, změňte >= před číslem verze na ==. Vyřešené verze každého balíčku jsou pak uloženy v uv.lock. Soubor lockfile zajišťuje, aby vývojáři pracující na projektu používali konzistentní verze balíčků. Také zajišťuje, aby se při distribuci balíčku koncovým uživatelům používala úplně stejná sada verzí balíčků. Soubor uv.lock byste neměli upravovat.
```
[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.5.0",
    "python-dotenv>=1.1.1",
    "pyarrow>=19.0.0",
]
```

Aktualizujte popis, aby byl popisnější.

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

Uložte a zavřete soubor.

Aktualizace main.py

Otevřete soubor s názvem main.py. Měl by se podobat tomuto příkladu.

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

if __name__ == "__main__":
    main()

main.py Obsah nahraďte následujícími bloky kódu. Každý blok vychází z předchozího bloku a měl by být v main.py daném pořadí.

Návod

Pokud má Visual Studio Code potíže s řešením balíčků, musíte interpreta aktualizovat tak, aby používal virtuální prostředí.

V horní části souboru main.pypřidejte importy a konstanty. Skript používá mssql_python pro připojení k databázi a načítání pomocí Arrow, pyarrow a pyarrow.parquet pro práci se sloupcovými daty a vstupně-výstupní operace se soubory Parquet, python-dotenv pro načítání připojovacích řetězců ze souboru .env a zkompilovaný vzor regulárního výrazu, který ověřuje identifikátory SQL, aby se zabránilo injekci.

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

import os
import re
import time

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

Pod importy přidejte mapování typu SQL-to-Arrow. Tento slovník překládá typy sloupců SQL Serveru na jejich ekvivalenty Apache Arrow, aby se při zápisu do Parquet zachovala věrnost dat. Pomocné funkce vytvářejí z metadat INFORMATION_SCHEMA přesné řetězce typů SQL (například NVARCHAR(100) nebo DECIMAL(18,2)) a pro každý sloupec určují odpovídající typ Arrow. Tyto typy jsou uloženy jako metadata polí v souborech Parquet, aby bylo možné cílovou tabulku znovu vytvořit s přesnými definicemi sloupců.

_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("us"),
    "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()

Přidejte introspekci schématu a funkce generování DDL. _get_arrow_schema INFORMATION_SCHEMA.COLUMNS provádí dotazy využívající parametrizované dotazy, sestaví Arrow schéma a uloží původní typ SQL jako metadata polí, aby bylo možné cílovou tabulku znovu vytvořit s přesnými definicemi sloupců. _create_table_ddl čte zpět tato metadata k generování DROP/CREATE TABLE DDL. Typ timestamp (rowversion) je namapován na VARBINARY(8) , protože se automaticky generuje a není možné ho přímo vložit.

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

Přidejte funkci pro stažení. download_table používá cursor.arrow_batch() k načítání dat přímo jako dávek záznamů Arrow ve vrstvě ovladače v jazyce C++, což zabraňuje vytváření mezilehlých objektů Pythonu. Každá dávka je převedena na schéma obohacené o metadata z _get_arrow_schema, aby původní typy SQL (například NVARCHAR(100)) byly zachovány v souboru Parquet. Funkce používá dva samostatné kurzory: jeden ke čtení metadat sloupců a druhý pro streamování dat.

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(schema_name)
    _validate_ident(table_name)
    source = f"{schema_name}.[{table_name}]"

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

    row_count = 0
    t0 = time.perf_counter()

    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {source}")
        writer = None
        try:
            while True:
                batch = cursor.arrow_batch(BATCH_SIZE)
                if batch.num_rows == 0:
                    break
                # Cast to the schema to preserve SQL type metadata in Parquet
                arrays = [
                    batch.column(i).cast(schema.field(i).type)
                    for i in range(batch.num_columns)
                ]
                batch = pa.record_batch(arrays, schema=schema)
                if writer is None:
                    writer = pq.ParquetWriter(parquet_file, schema)
                writer.write_batch(batch)
                row_count += batch.num_rows
        finally:
            if writer is not None:
                writer.close()

    if row_count == 0:
        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

Přidejte háček pro rozšiřování. enrich_parquet je zástupný symbol, ve kterém můžete před nahráním přidat transformace, odvozené sloupce nebo spojení k datům. V tomto rychlém startu se jedná o no-op, který vrátí cestu k souboru beze změny.

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

Přidejte funkci pro nahrání. upload_parquet přečte Arrow schema ze souboru Parquet, vygeneruje a spustí DROP/CREATE TABLE DDL pro přípravu cíle, poté načte soubor v dávkách a využije cursor.bulkcopy() pro vysoce výkonné hromadné vložení. Tato table_lock=True možnost zlepšuje propustnost minimalizací kolizí zámků. Po dokončení nahrávání funkce spustí SELECT COUNT(*) a vyvolá chybu, pokud počet cílových řádků neodpovídá počtu nahraných řádků.

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
    elapsed = time.perf_counter() - t0

    # ── Verify ──
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT COUNT(*) FROM {target}")
        count = cursor.fetchone()[0]
    if count != uploaded:
        raise ValueError(
            f"Row count mismatch for {target}: uploaded {uploaded:,}, destination has {count:,}"
        )

    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"| destination rows: {count:,}"
    )
    return uploaded

Přidejte funkci orchestrace. transfer_tables spojí tři fáze dohromady. Připojí se ke zdrojové databázi, objeví všechny základní tabulky v daném schématu pomocí INFORMATION_SCHEMA.TABLES, stáhne každou do lokálního souboru Parquet, spustí obohacovací hák, poté se připojí k cílové databázi a nahraje každý soubor.

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 = []
    for table_name, parquet_file in parquet_files:
        enriched.append((table_name, enrich_parquet(parquet_file)))

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

Nakonec přidejte vstupní bod main. Načte .env soubor, volá transfer_tables se zdrojovými a cílovými připojovacími řetězci a vytiskne celkový uplynulý čas.

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

Uložte a zavřete main.py.

Uložte připojovací řetězce

.gitignore Otevřete soubor a přidejte vyloučení souborů.env. Soubor by měl být podobný tomuto příkladu. Až budete hotovi, nezapomeňte ho uložit a zavřít.
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env
```
V aktuálním adresáři vytvořte nový soubor s názvem .env.
.env V souboru přidejte položky pro zdrojové a cílové připojovací řetězce. Nahraďte zástupné hodnoty skutečnými názvy serverů a databází.
```
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"
```
Návod

Připojovací řetězec použitý zde do značné míry závisí na typu databáze SQL, ke které se připojujete. Pokud se připojujete k Azure SQL Database nebo k SQL databázi ve Fabric, použijte připojovací řetězec ODBC z karty Připojovací řetězce. Možná budete muset upravit typ ověřování v závislosti na vašem scénáři. Další informace o připojovacích řetězcích a jejich syntaxi najdete v referenčních informacích k syntaxi připojovacího řetězce.

Návod

V systému macOS funguje ověřování pomocí Microsoft Entra jak s ActiveDirectoryInteractive, tak s ActiveDirectoryDefault. ActiveDirectoryInteractive vás vyzve k přihlášení při každém spuštění skriptu. Abychom se vyhnuli opakovaným výzvám k přihlášení, přihlaste se jednou přes Azure CLI spuštěním az login, poté použijte ActiveDirectoryDefault, které znovu použije cacheované přihlašovací údaje.

Použijte 'uv run' ke spuštění skriptu

V okně terminálu před nebo v novém okně terminálu, které se otevře ve stejném adresáři, spusťte následující příkaz.

 uv run main.py

Tady je očekávaný výstup po dokončení skriptu.

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

Připojte se k cílové databázi pomocí rozšíření MSSQL pro VS Code a ověřte, že tabulky a data byly úspěšně vytvořeny.
Pokud chcete skript nasadit do jiného počítače, zkopírujte všechny soubory s výjimkou .venv složky na druhý počítač. Virtuální prostředí se znovu vytvoří při prvním spuštění.

Jak kód funguje

Aplikace provádí kompletní přenos dat během zpáteční cesty ve třech fázích:

Ke stažení: Připojí se ke zdrojové databázi, přečte metadata sloupců z INFORMATION_SCHEMA.COLUMNS, sestaví schéma Apache Arrow a pak stáhne každou tabulku do místního souboru Parquet.
Obohacení (volitelné): Poskytuje háček (enrich_parquet), kde můžete před nahráním přidat transformace, odvozené sloupce nebo spojení.
Nahrání: Načte každý soubor Parquet v dávkách, znovu vytvoří tabulku v cílové databázi pomocí DDL generovaného z metadat schématu Arrow a použije cursor.bulkcopy() k výkonnému hromadnému vložení.

Další kroky

Použijte tyto články k dalšímu budování:

Hromadná kopie pro pokročilé hromadné kopírovací vzory včetně mapování sloupců, velikostování šarží a zpracování chyb.
Načítání a pohybové vzory dat pro strategie pro načítání dat ze souborů, API a dalších databází.
Ladění výkonu pro optimalizaci propustnosti pro operace s velkými daty.

hromadné kopírování

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-22