Быстрый старт: Массовое копирование с использованием драйвера mssql-python для Python

В этом кратком руководстве вы используете mssql-python драйвер для пакетного копирования данных между базами данных. Приложение загружает таблицы из схемы исходной базы данных в локальные файлы Parquet с помощью Apache Arrow, а затем отправляет их в целевую базу данных с помощью высокопроизводительного bulkcopy метода. Этот шаблон можно использовать для переноса, репликации или преобразования данных между SQL Server, базой данных SQL Azure и базой данных SQL в Fabric.

Драйвер mssql-python не требует внешних зависимостей на компьютерах Windows. Драйвер устанавливает все, что требуется с одной pip установкой, что позволяет использовать самую последнюю версию драйвера в новых сценариях без нарушения работы других сценариев, для обновления и тестирования которых у вас нет времени.

Документация | Исходный код mssql-python | Пакет (PyPI) | uv

Необходимые условия

Python 3.10 или более поздней версии
Если у вас еще нет Python, установите среду выполнения Python и диспетчер пакетов pip из python.org.
Не хотите использовать собственную среду? Следуйте инструкциям в разделе Контейнерная и локальная разработка, чтобы создать воспроизводимую среду разработки в devcontainer или GitHub Codespaces.
Visual Studio Code со следующими расширениями:
- Расширение Python для Visual Studio Code
Интерфейс azure Command-Line (CLI) для проверки подлинности без пароля в macOS и Linux.
Если у вас еще нет uv, следуйте инструкциям по установке.
Исходная база данных со схемой-примером AdventureWorksLT и действительной строкой подключения.
База данных назначения с действительной строкой подключения. Пользователь должен иметь разрешение на создание и запись в таблицы. Если у вас нет второй базы данных, вы можете использовать ту же базу данных и другую схему для таблиц назначения.

Установите единовременные предварительные условия для операционной системы. Пользователи Windows могут пропустить этот шаг. Для полной информации о платформе см. Установить 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

Создание базы данных SQL

Создайте или подключитесь к SQL-базе данных на одной из следующих платформ:

Создание базы данных SQL в минутах с помощью портала Azure

Создание проекта и запуск кода

Создание нового проекта
Добавление зависимостей
Запуск Visual Studio Code
Обновление pyproject.toml
Обновление main.py
Сохранение строк подключения
Использование uv run для выполнения скрипта

Создание нового проекта

Откройте командную строку в каталоге разработки. Если у вас его нет, создайте новую директорию, например python или scripts. Избегайте папок на OneDrive, так как синхронизация может мешать управлению виртуальной средой.

Создайте новый проект с uv.

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

Добавление зависимостей

В том же каталоге установите mssql-pythonи python-dotenvpyarrow пакеты.

uv add mssql-python python-dotenv pyarrow

Запустите Visual Studio Code.

В том же каталоге выполните следующую команду.

code .

Обновление pyproject.toml

Pyproject.toml содержит метаданные проекта. Откройте файл в избранном редакторе.
Просмотрите содержимое файла. Он должен быть похож на этот пример. Обратите внимание на версию Python и зависимость: mssql-python использует >= для определения минимальной версии. Если вы предпочитаете точную версию, измените >= на == перед номером версии. Затем разрешенные версии каждого пакета хранятся в uv.lock. Lockfile гарантирует, что разработчики, работающие над проектом, используют согласованные версии пакетов. Он также гарантирует, что при распространении пакета пользователям используется тот же набор версий пакетов. Не следует изменять 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.5.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 следующими блоками кода. Каждый блок строится на предыдущем и должен быть размещен в main.py последовательно.

Подсказка

Если в Visual Studio Code возникли проблемы с разрешением пакетов, необходимо обновить интерпретатор для использования виртуальной среды.
В верхней части main.pyдобавьте импорт и константы. Скрипт использует mssql_python для подключения к базе данных и извлечения данных Arrow, pyarrow и pyarrow.parquet для обработки колоночных данных и ввода-вывода файлов Parquet, 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

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-to-Arrow. Этот словарь преобразует типы столбцов SQL Server в эквивалентные типы Apache Arrow, чтобы сохранялась точность данных при записи в Parquet. Вспомогательные функции строят точные строки SQL-типов (например NVARCHAR(100) , или DECIMAL(18,2)) из INFORMATION_SCHEMA метаданных и определяют соответствующий тип стрелки для каждого столбца. Эти типы хранятся в виде полевых метаданных в файлах Parquet, чтобы таблица назначения могла быть воссоздана с точными определениями столбцов.

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

Добавьте функции интроспекции схемы и генерации 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 использует cursor.arrow_batch() для непосредственного получения данных в виде пакетов записей Arrow на уровне драйвера, написанного на C++, что позволяет избежать создания промежуточных объектов Python. Каждая партия приводится к схеме, обогащённой метаданными, из _get_arrow_schema, чтобы исходные SQL-типы (например, NVARCHAR(100)) сохранялись в файле Parquet. Функция использует два отдельных курсора: один для чтения метаданных столбца, а другой — для потоковой передачи данных.

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

Добавьте крючок обогащения. 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, or joins
    print(f"Enriching {parquet_file} (no-op)")
    return parquet_file
```

Добавьте функцию отправки. upload_parquet считывает схему Arrow из файла Parquet, создает и выполняет DROP/CREATE TABLE DDL для подготовки назначения, затем считывает файл пакетами и вызывает cursor.bulkcopy() для высокопроизводительной массовой вставки. Этот table_lock=True параметр повышает пропускную способность путем минимизации конфликтов блокировки. После завершения загрузки функция выполняет 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
    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

Добавьте функцию оркестрации. 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 = []
    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)

Наконец, добавьте точку 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"
```
Подсказка

Строка подключения, используемая здесь, в значительной степени зависит от типа базы данных SQL, к которой вы подключаетесь. Если вы подключаетесь к базе данных SQL Azure или базе данных SQL в Fabric, используйте строку подключения ODBC на вкладке строк подключения. Возможно, вам потребуется настроить тип проверки подлинности в зависимости от вашего сценария. Дополнительные сведения о строках подключения и их синтаксисе см. в справочнике по синтаксису строки подключения.

Подсказка

На macOS, оба ActiveDirectoryInteractive и ActiveDirectoryDefault работают для проверки подлинности Microsoft Entra. ActiveDirectoryInteractive запрашивает вход при каждом запуске скрипта. Чтобы избежать повторных запросов при входе, войдите один раз через Azure CLI, запустив az login, затем используйте ActiveDirectoryDefault, который повторно использует кэшированный учетный код.

Использование 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
```
Подключитесь к базе данных получателя, используя расширение MSSQL для VS Code, и убедитесь, что таблицы и данные были успешно созданы.
Чтобы развернуть скрипт на другом компьютере, скопируйте все файлы, кроме .venv папки на другой компьютер. Виртуальная среда повторно создается при первом запуске.

Как работает код

Приложение выполняет полную передачу данных по пути в три этапа:

Скачать: подключается к исходной базе данных, считывает метаданные столбца из INFORMATION_SCHEMA.COLUMNS, создает схему Apache Arrow, а затем загружает каждую таблицу в локальный файл Parquet.
Обогащение (необязательно): предоставляет точку расширения (enrich_parquet), где можно добавлять преобразования, производные столбцы и объединения перед отправкой.
Отправка: считывает каждый файл Parquet пакетами, воссоздает таблицу в целевой базе данных с использованием DDL, созданной из метаданных схемы Apache Arrow, а затем использует cursor.bulkcopy() для высокой производительности массовой вставки.

Дальнейшие действия

Используйте эти статьи, чтобы продолжать строить:

Массовое копирование для расширенных сценариев массового копирования, включая сопоставление столбцов, размер пакета и обработку ошибок.
Загрузка и паттерны перемещения данных для стратегий загрузки данных из файлов, API и других баз данных.
Настройка производительности для оптимизации пропускной способности для больших операций с данными.

Массовое копирование

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-07-22