Usa mssql-python con DuckDB

DuckDB es un motor de análisis SQL en proceso que puede consultar directamente las tablas Apache Arrow sin copiar datos. Combinar DuckDB con el controlador mssql-python te permite:

  • Ejecuta consultas SQL analíticas en conjuntos de resultados de Microsoft SQL sin cargar datos en pandas o Polars.
  • Consulta tablas Arrow en memoria sin sobrecarga de copia.
  • Únete los datos SQL de Microsoft con archivos locales (CSV, Parquet, JSON) en una sola consulta de DuckDB.
  • Exporta datos SQL de Microsoft a Parquet, CSV u otros formatos a través de DuckDB.

Prerequisites

  • Python 3.10 o posterior.
  • Los paquetes mssql-python, duckdb y pyarrow. Instala todo con pip install mssql-python duckdb pyarrow.
  • Instale requisitos previos específicos del sistema operativo de un solo uso. Los usuarios de Windows pueden saltarse este paso. Para detalles completos sobre la plataforma, véase Instalar mssql-python.
    apk add libtool krb5-libs krb5-dev

Creación de una base de datos SQL

Crea o conéctate a una base de datos SQL en una de las siguientes plataformas:

Los ejemplos de este artículo consultan la AdventureWorks base de datos de ejemplo. Si aún no lo tienes, consulta las bases de datos de ejemplo de AdventureWorks.

Instalación de dependencias

pip install mssql-python duckdb pyarrow

Consulta datos SQL de Microsoft con DuckDB

El flujo de trabajo básico es: ejecutar una consulta con mssql-python, obtener los resultados como una tabla de flechas y luego consultar esa tabla de flechas con DuckDB SQL.

Patrón básico

Empieza por establecer una conexión y obtener los datos como una tabla 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())

DuckDB hace referencia a la products tabla de flechas por su nombre de variable en Python. No se copian datos en el almacenamiento de DuckDB.

Agregar y filtrar

Usa SQL de DuckDB para agrupar y agregar datos de 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())

Unir múltiples resultados de Microsoft SQL

Obtén varias tablas de Microsoft SQL y únelas en DuckDB sin escribir una 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 datos SQL de Microsoft con archivos locales

DuckDB puede leer archivos CSV, Parquet y JSON de forma nativa. Combina los datos de SQL Server con archivos locales en una sola consulta.

Unir con un archivo CSV

Carga un archivo CSV y únelo con datos de 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)

Unirse a un archivo Parquet

Carga un archivo Parquet y únelo con datos de 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 datos SQL de Microsoft

Utiliza la declaración de COPY DuckDB para exportar datos de Microsoft SQL a varios formatos de archivo.

Exportar a Parquet

Exportar datos al formato Apache Parquet.

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

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

Exportar a CSV

Exportar datos a un archivo de valores separados por comas:

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

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

Parquet particionado para exportación

Exportar datos a archivos Parquet particionados para análisis distribuido:

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 en flujo grandes conjuntos de resultados

Para conjuntos de datos grandes, úsase arrow_reader() para procesar datos en lotes de streaming sin cargar todas las filas en memoria a la 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 entre todos los lotes, registre cada lote en una conexión persistente de DuckDB y acumule los resultados de manera 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()

Consejos de rendimiento

Deja que Microsoft SQL se encargue del trabajo pesado

Microsoft SQL es más rápido para filtrar, combinar y realizar agregaciones que transferir todos los datos sin procesar a través de la red. Utiliza DuckDB para análisis secundario de conjuntos de resultados que ya están obtenidos, no como sustituto de la optimización de consultas de 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")

Usa Arrow para todas las operaciones de lectura

La transferencia basada en flechas evita crear objetos Python intermedios, lo que reduce el consumo de memoria y mejora el rendimiento. Prefiera cursor.arrow() a la conversión manual, fila por fila, al pasar datos a DuckDB.

Utiliza el streaming para grandes conjuntos de datos

Para conjuntos de resultados mayores que la memoria disponible, úsate arrow_reader() con un batch_size parámetro para procesar datos de forma incremental.

Contenido relacionado

  • Last updated on