Use mssql-python com DuckDB

DuckDB é um motor de análise SQL em processo que pode consultar tabelas Apache Arrow diretamente sem copiar dados. Combinar o DuckDB com o driver mssql-python permite que você:

  • Execute consultas analíticas SQL em conjuntos de resultados do Microsoft SQL sem carregar dados no pandas ou no Polars.
  • Consulte tabelas Arrow na memória sem overhead de cópia.
  • Junte dados SQL da Microsoft com arquivos locais (CSV, Parquet, JSON) em uma única consulta DuckDB.
  • Exporte dados SQL da Microsoft para Parquet, CSV ou outros formatos via DuckDB.

Pré-requisitos

  • Python 3.10 ou posterior.
  • Os pacotes mssql-python, duckdb e pyarrow. Instale tudo com pip install mssql-python duckdb pyarrow.
  • Instale pré-requisitos específicos do sistema operacional uma vez. Usuários do Windows podem pular essa etapa. Para detalhes completos da plataforma, veja Instalar mssql-python.
    apk add libtool krb5-libs krb5-dev

Criar um banco de dados SQL

Crie ou conecte-se a um banco de dados SQL em uma das seguintes plataformas:

Os exemplos neste artigo consultam o AdventureWorks banco de dados de exemplo. Se você ainda não tem, veja bancos de dados de exemplo do AdventureWorks.

Instalar dependências

pip install mssql-python duckdb pyarrow

Consultar dados SQL da Microsoft com o DuckDB

O fluxo básico de trabalho é: execute uma consulta com mssql-python, busque os resultados como uma tabela de setas e depois consulte essa tabela de setas com DuckDB SQL.

Padrão básico

Primeiro, estabeleça uma conexão e obtenha os dados como uma tabela Arrow.

import duckdb
import mssql_python

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes"
)
cursor = conn.cursor()

# Fetch Microsoft SQL data as Arrow
cursor.execute("SELECT * FROM Production.Product WHERE ListPrice > 0")
products = cursor.arrow()

# Query the Arrow table with DuckDB
result = duckdb.sql("""
    SELECT Color, COUNT(*) AS ProductCount, AVG(ListPrice) AS AvgPrice
    FROM products
    GROUP BY Color
    ORDER BY ProductCount DESC
""")
print(result.fetchdf())

O DuckDB faz referência à tabela Arrow products pelo nome da variável Python. Nenhum dado é copiado para o armazenamento do DuckDB.

Agregar e filtrar

Use o SQL do DuckDB para agrupar e agregar dados do Arrow.

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

# Top customers by total spend
top_customers = duckdb.sql("""
    SELECT
        CustomerID,
        COUNT(*) AS OrderCount,
        SUM(TotalDue) AS TotalSpent,
        AVG(TotalDue) AS AvgOrderValue
    FROM orders
    GROUP BY CustomerID
    HAVING SUM(TotalDue) > 10000
    ORDER BY TotalSpent DESC
    LIMIT 20
""")
print(top_customers.fetchdf())

Combine vários resultados do Microsoft SQL

Busque várias tabelas do Microsoft SQL e junte-as no DuckDB sem escrever uma consulta entre servidores.

# Fetch two tables
cursor.execute("SELECT * FROM Production.Product")
products = cursor.arrow()

cursor.execute("SELECT * FROM Production.ProductSubcategory")
subcategories = cursor.arrow()

# Join in DuckDB
result = duckdb.sql("""
    SELECT
        s.Name AS Subcategory,
        COUNT(*) AS ProductCount,
        ROUND(AVG(p.ListPrice), 2) AS AvgPrice
    FROM products p
    JOIN subcategories s ON p.ProductSubcategoryID = s.ProductSubcategoryID
    GROUP BY s.Name
    ORDER BY AvgPrice DESC
""")
print(result.fetchdf())

Unir dados SQL da Microsoft com arquivos locais

O DuckDB pode ler arquivos CSV, Parquet e JSON nativamente. Combine dados do SQL Server com arquivos locais em uma única consulta.

Unir com um arquivo CSV

Carregue um arquivo CSV e junte-o com dados do Microsoft SQL.

import csv
from pathlib import Path

cursor.execute("SELECT CustomerID, PersonID FROM Sales.Customer")
customers = cursor.arrow()

csv_path = Path("customer_regions.csv")
with csv_path.open("w", newline="", encoding="utf-8") as file:
    writer = csv.writer(file)
    writer.writerow(["CustomerID", "Region", "Segment"])
    writer.writerows([
        (1, "West", "Premium"),
        (2, "East", "Standard"),
        (3, "Central", "Basic"),
    ])

try:
    result = duckdb.sql("""
        SELECT c.CustomerID, c.PersonID, f.Region, f.Segment
        FROM customers c
        JOIN read_csv_auto('customer_regions.csv') f ON c.CustomerID = f.CustomerID
    """)
    print(result.fetchdf())
finally:
    csv_path.unlink(missing_ok=True)

Fazer junção com um arquivo Parquet

Carregue um arquivo Parquet e combine-o com dados do SQL Server da Microsoft.

from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
products = cursor.arrow()

parquet_path = Path("order_history.parquet")
order_history = pa.table({
    "ProductID": [1, 2, 680],
    "OrderDate": ["2024-06-01", "2024-03-15", "2024-01-10"],
    "Quantity": [10, 5, 3],
})
pq.write_table(order_history, parquet_path)

try:
    result = duckdb.sql("""
        SELECT p.Name, p.ListPrice, h.OrderDate, h.Quantity
        FROM products p
        JOIN read_parquet('order_history.parquet') h ON p.ProductID = h.ProductID
        WHERE h.OrderDate >= '2024-01-01'
    """)
    print(result.fetchdf())
finally:
    parquet_path.unlink(missing_ok=True)

Exportar dados SQL da Microsoft

Use a instrução do COPY DuckDB para exportar dados do Microsoft SQL para vários formatos de arquivo.

Exportar para Parquet

Exportar dados para o formato Apache Parquet.

cursor.execute("SELECT * FROM Production.Product")
products = cursor.arrow()

duckdb.sql("COPY products TO 'products.parquet' (FORMAT PARQUET)")

Exportar para CSV

Exporte dados para um arquivo de valores separados por vírgulas:

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

duckdb.sql("COPY orders TO 'orders.csv' (FORMAT CSV, HEADER)")

Parquet particionado para exportação

Exporte dados para arquivos Parquet particionados para análise distribuída:

import shutil
from pathlib import Path

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

output_dir = Path("sales_data")
shutil.rmtree(output_dir, ignore_errors=True)

duckdb.sql("""
    COPY (SELECT *, YEAR(OrderDate) AS OrderYear FROM orders)
    TO 'sales_data'
    (FORMAT PARQUET, PARTITION_BY (OrderYear))
""")

Transmitir grandes conjuntos de resultados em fluxo

Para grandes conjuntos de dados, use arrow_reader() para processar dados em lotes de streaming sem carregar todas as linhas na memória de uma vez:

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=50000)

# Process each batch with DuckDB
total_rows = 0
for batch in reader:
    result = duckdb.sql("""
        SELECT ProductID, SUM(ActualCost) AS TotalCost
        FROM batch
        GROUP BY ProductID
    """)
    total_rows += batch.num_rows
    print(f"Processed {total_rows} rows")

Acumular resultados de transmissão

Para consolidar os resultados de todos os lotes, registre cada lote em uma conexão persistente com o DuckDB e acumule os resultados de forma incremental.

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=50000)

duck = duckdb.connect()
duck.execute("CREATE TABLE transactions (ProductID INT, ActualCost DOUBLE, Quantity INT)")

for batch in reader:
    duck.execute("INSERT INTO transactions SELECT ProductID, ActualCost, Quantity FROM batch")

# Query the accumulated data
result = duck.sql("""
    SELECT ProductID, SUM(ActualCost) AS TotalCost, SUM(Quantity) AS TotalQty
    FROM transactions
    GROUP BY ProductID
    ORDER BY TotalCost DESC
    LIMIT 10
""")
print(result.fetchdf())
duck.close()

Dicas de desempenho

Deixe o Microsoft SQL cuidar do trabalho pesado

O Microsoft SQL é mais rápido para filtragem, junções e agregações do que transferir todos os dados brutos pela rede. Use o DuckDB para análise secundária em conjuntos de resultados que já foram buscados, não como substituto da otimização de consultas do SQL Server.

# Suboptimal: Pull all rows, filter in DuckDB
cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()
result = duckdb.sql("SELECT * FROM orders WHERE TotalDue > 1000")

# Better: Filter in Microsoft SQL, analyze in DuckDB
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE TotalDue > 1000")
orders = cursor.arrow()
result = duckdb.sql("SELECT CustomerID, SUM(TotalDue) FROM orders GROUP BY CustomerID")

Use o Arrow para todas as operações de leitura

A transferência baseada em seta evita criar objetos Python intermediários, o que reduz o uso de memória e melhora a taxa de transferência. Prefira cursor.arrow() à conversão manual linha por linha ao passar dados para o DuckDB.

Use streaming para grandes conjuntos de dados

Para conjuntos de resultados maiores que a memória disponível, use arrow_reader() com um batch_size parâmetro para processar dados incrementalmente.

Conteúdo relacionado

  • Last updated on