Usa mssql-python com DuckDB

O 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-lhe:

  • Executar consultas analíticas SQL sobre conjuntos de resultados do Microsoft SQL sem carregar dados para o pandas ou o Polars.
  • Consultar tabelas Arrow em memória sem overhead de cópia.
  • Junta dados SQL da Microsoft com ficheiros locais (CSV, Parquet, JSON) numa única consulta DuckDB.
  • Exporte dados SQL da Microsoft para Parquet, CSV ou outros formatos através do 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 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

Criar um banco de dados SQL

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

Os exemplos deste artigo consultam a AdventureWorks base de dados de exemplo. Se ainda não o tens, vê as bases 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 de trabalho básico é: executar uma consulta com mssql-python, obter os resultados como uma tabela de setas e depois consultar essa tabela de setas com DuckDB SQL.

Padrão básico

Comece por estabelecer uma ligação e obter 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 refere a products tabela Arrow pelo nome da sua variável em 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

Buscar várias tabelas do Microsoft SQL e juntá-las 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())

Juntar dados Microsoft SQL com ficheiros locais

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

Juntar com um ficheiro CSV

Carrega um ficheiro CSV e junta-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)

Juntar a um ficheiro Parquet

Carrega um ficheiro Parquet e associe-o a dados do Microsoft SQL.

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 declaração do COPY DuckDB para exportar dados Microsoft SQL para vários formatos de ficheiro.

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

Exportar dados para um ficheiro 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

Exportar dados para ficheiros 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))
""")

Transmita grandes conjuntos de resultados

Para conjuntos de dados grandes, use arrow_reader() para processar dados em lotes de streaming sem carregar todas as linhas na memória de uma só 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 streaming

Para agregar os resultados de todos os lotes, registe cada lote numa ligação persistente ao DuckDB e acumule os resultados incrementalmente.

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

Sugestões de desempenho

Deixe o Microsoft SQL tratar do trabalho pesado

O SQL da Microsoft é mais rápido na filtragem, nas junções e nas agregações do que extrair todos os dados brutos através da rede. Use o DuckDB para análise secundária de conjuntos de resultados já obtidos, 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 Arrow evita a criação de objetos Python intermédios, o que reduz a utilização de memória e melhora a taxa de transferência. Prefiro cursor.arrow() em vez de conversão manual linha a linha ao passar dados para o DuckDB.

Use streaming para grandes conjuntos de dados

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

Conteúdo relacionado

  • Last updated on