Início rápido: Cópia em massa com o driver mssql-python para Python

Neste guia de início rápido, utilizarás o driver mssql-python para copiar dados em massa entre bases de dados. A aplicação descarrega tabelas de um esquema de base de dados de origem para ficheiros Parquet locais usando o Apache Arrow, e depois carrega-as para uma base de dados de destino usando o método de alto desempenho bulkcopy . Pode usar este padrão para migrar, replicar ou transformar dados entre SQL Server, Base de Dados SQL do Azure e SQL Database no Fabric.

O mssql-python driver não requer nenhuma dependência externa em máquinas Windows. O driver instala tudo o que precisa com uma única pip instalação, permitindo que você use a versão mais recente do driver para novos scripts sem quebrar outros scripts que você não tem tempo para atualizar e testar.

Documentação | Código-fonte | Pacote (PyPI) | UV

Pré-requisitos

Python 3.10 ou posterior
Se ainda não tens o Python, instala o runtime do Python e o gestor de pacotes pip a partir de python.org.
Não quer usar o seu próprio ambiente? Consulte Container and local development para criar um ambiente devcontainer reproduzível ou um ambiente do GitHub Codespaces.
Visual Studio Code com as seguintes extensões:
- Extensão Python para Visual Studio Code
Azure Command-Line Interface (CLI) para autenticação sem palavra-passe no macOS e Linux.
Se ainda não tiver uv, siga as instruções de instalação.
Uma base de dados de origem com o AdventureWorksLT esquema de exemplo e uma cadeia de ligação válida.
Uma base de dados de destino com uma cadeia de ligação válida. O utilizador deve ter permissão para criar e escrever em tabelas. Se não tiveres uma segunda base de dados, podes usar a mesma base de dados e um esquema diferente para as tabelas de destino.

Instale pré-requisitos que devem ser instalados uma única vez específicos do sistema operacional. Os utilizadores do Windows podem saltar este passo. Para detalhes completos da plataforma, consulte Instalar 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

Criar um banco de dados SQL

Criar ou ligar-se a uma base de dados SQL numa das seguintes plataformas:

Criar um banco de dados SQL em minutos usando o portal do Azure

Criar um novo projeto

Abra um prompt de comando no diretório de desenvolvimento. Se não tiver um, crie um novo diretório, como python ou scripts. Evite pastas no seu OneDrive, pois a sincronização pode interferir na gestão do seu ambiente virtual.

Crie um novo projeto com uv.

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

Adicionar dependências

No mesmo diretório, instale os pacotes mssql-python, python-dotenv e pyarrow.

uv add mssql-python python-dotenv pyarrow

Abra o Visual Studio Code.

No mesmo diretório, execute o seguinte comando.

code .

Atualizar pyproject.toml

O pyproject.toml contém os metadados para o seu projeto. Abra o arquivo em seu editor favorito.
Revise o conteúdo do arquivo. Deve ser semelhante a este exemplo. Nota a versão do Python e a dependência para mssql-python usa >= para definir uma versão mínima. Se preferir uma versão exata, altere o >= para == antes do número da versão. As versões resolvidas de cada pacote são então armazenadas no uv.lock. O arquivo de bloqueio garante que os desenvolvedores que trabalham no projeto estejam usando versões de pacote consistentes. Ele também garante que o mesmo conjunto exato de versões de pacote seja usado ao distribuir seu pacote para usuários finais. Você não deve editar o uv.lock arquivo.
```
[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",
]
```

Atualize a descrição para ser mais descritiva.

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

Salve e feche o arquivo.

Atualizar main.py

Abra o arquivo chamado main.py. Deve ser semelhante a este exemplo.

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

if __name__ == "__main__":
    main()

Substitua o conteúdo de main.py pelos seguintes blocos de código. Cada bloco constrói-se sobre o anterior e deve ser colocado main.py na ordem correta.

Sugestão

Se o Visual Studio Code estiver tendo problemas para resolver pacotes, você precisará atualizar o intérprete para usar o ambiente virtual.

No topo de main.py, adicione as importações e as constantes. O script utiliza mssql_python para conectividade com base de dados e busca por setas, pyarrow e pyarrow.parquet para tratamento de dados colunares e I/O de ficheiros Parquet, python-dotenv para carregar strings de ligação a partir de um .env ficheiro, e um padrão regex compilado que valida identificadores SQL para evitar injeção.

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

Por baixo das importações, adicione o mapeamento do tipo SQL-para-Arrow. Este dicionário traduz os tipos de colunas do SQL Server para os seus equivalentes no Apache Arrow, de modo a preservar a fidelidade dos dados ao escrever no Parquet. As funções auxiliares constroem cadeias exatas de tipos SQL (por exemplo, NVARCHAR(100) ou DECIMAL(18,2)) a partir dos metadados de INFORMATION_SCHEMA e determinam o tipo Arrow correspondente para cada coluna. Estes tipos são armazenados como metadados de campo nos ficheiros Parquet para que a tabela de destino possa ser recriada com definições exatas de colunas.

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

Adiciona as funções de introspeção de esquemas e geração DDL. _get_arrow_schema consultas INFORMATION_SCHEMA.COLUMNS usando consultas parametrizadas, constrói um esquema Arrow e armazena o tipo SQL original como metadados de campo para que a tabela de destino possa ser recriada com definições exatas de colunas. _create_table_ddl lê esses metadados para gerar DROP/CREATE TABLE DDL. O tipo timestamp (rowversion) é remapeado para VARBINARY(8) porque é gerado automaticamente e não é inserido diretamente.

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

Adiciona a função de download. download_table usa cursor.arrow_batch() para obter dados diretamente sob a forma de lotes de registos Arrow na camada C++ do driver, o que evita a criação de objetos Python intermédios. Cada lote é convertido para o esquema enriquecido com metadados de _get_arrow_schema, para que os tipos SQL originais (por exemplo, NVARCHAR(100)) sejam preservados no ficheiro Parquet. A função utiliza dois cursores separados: um para ler metadados das colunas e outro para transmitir os dados.

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

Adicione o gancho de enriquecimento. enrich_parquet é um marcador onde podes adicionar transformações, colunas derivadas ou junções aos dados antes de serem carregados. Neste quickstart, consiste numa operação nula que devolve o caminho do ficheiro inalterado.
```
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
```

Adiciona a função de upload. upload_parquet lê o esquema Arrow do ficheiro Parquet, gera e executa DROP/CREATE TABLE DDL para preparar o destino, após lê o ficheiro em lotes e chama cursor.bulkcopy() para uma inserção em massa de alto desempenho. A table_lock=True opção melhora o rendimento ao minimizar a contenção de bloqueios. Depois de o upload terminar, a função executa um SELECT COUNT(*) e gera um erro se a contagem de linhas de destino não corresponder à contagem carregada.

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

Adiciona a função de orquestração. transfer_tables Liga as três fases. Liga-se à base de dados de origem, descobre todas as tabelas de base no esquema indicado através de INFORMATION_SCHEMA.TABLES, descarrega cada uma delas para um ficheiro Parquet local, executa o gancho de enriquecimento e, em seguida, liga-se à base de dados de destino e carrega cada ficheiro.

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)

Por fim, adiciona o ponto de entrada main. Carrega o .env ficheiro, chama transfer_tables com as cadeias de ligação de origem e destino, e imprime o tempo total decorrido.

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

Salve e feche main.py.

Guardar as cadeias de ligação

Abra o .gitignore arquivo e adicione uma exclusão para .env arquivos. Seu arquivo deve ser semelhante a este exemplo. Certifique-se de salvá-lo e fechá-lo quando terminar.
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env
```
No diretório atual, crie um novo arquivo chamado .env.
Dentro do .env ficheiro, adiciona entradas para as cadeias de ligação de origem e destino. Substitui os valores provisórios pelos nomes reais do teu servidor e base de dados.
```
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"
```
Sugestão

A cadeia de conexão usada aqui depende em grande parte do tipo de banco de dados SQL ao qual você está se conectando. Se você estiver se conectando a um Banco de Dados SQL do Azure ou a um banco de dados SQL na Malha, use a cadeia de conexão ODBC na guia Cadeias de conexão. Talvez seja necessário ajustar o tipo de autenticação dependendo do cenário. Para obter mais informações sobre cadeias de conexão e sua sintaxe, consulte Referência de sintaxe de cadeia de conexão.

Sugestão

No macOS, tanto ActiveDirectoryInteractive como ActiveDirectoryDefault funcionam para a autenticação do Microsoft Entra. ActiveDirectoryInteractive Solicita-te para iniciar sessão sempre que executas o script. Para evitar pedidos repetidos para iniciar sessão, inicie sessão uma vez através da CLI do Azure ao executar az login, e depois utilize ActiveDirectoryDefault, que reutiliza a credencial em cache.

Use uv run para executar o script

Na janela do terminal anterior ou em uma nova janela do terminal aberta no mesmo diretório, execute o seguinte comando.

 uv run main.py

Aqui está a saída esperada quando o script for concluído.

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

Ligue-se à base de dados de destino usando a extensão MSSQL para o VS Code e verifique se as tabelas e dados foram criados com sucesso.
Para implantar o script em outra máquina, copie todos os arquivos, exceto a .venv pasta, para a outra máquina. O ambiente virtual é recriado com a primeira execução.

Como funciona o código

A aplicação realiza uma transferência completa de dados de ida e volta em três fases:

Download: Liga-se à base de dados de origem, lê metadados das colunas de INFORMATION_SCHEMA.COLUMNS, constrói um esquema Apache Arrow e depois descarrega cada tabela para um ficheiro Parquet local.
Enriquecer (opcional): Fornece um gancho (enrich_parquet) onde pode adicionar transformações, colunas derivadas ou junções antes de carregar.
Upload: Lê cada ficheiro Parquet em lotes, recria a tabela na base de dados de destino usando DDL gerado a partir dos metadados do esquema Arrow, e depois usa cursor.bulkcopy() para inserção massiva de alto desempenho.

Passos seguintes

Use estes artigos para continuar a construir:

Cópia em massa para padrões avançados de cópia em massa, incluindo mapeamentos de colunas, tamanho em lote e tratamento de erros.
Padrões de carregamento e movimento de dados para estratégias de carregamento de dados a partir de ficheiros, APIs e outras bases de dados.
Ajuste de desempenho para otimizar o throughput para operações de grandes volumes de dados.

Cópia em massa

Comentários

Esta página foi útil?

Last updated on 2026-07-22