Inicio rápido: Apache Arrow con el controlador mssql-python para Python

En esta guía de inicio rápido, usa los métodos integrados de recuperación de Arrow del controlador mssql-python para recuperar datos de SQL Server como tablas columnares de Apache Arrow. El formato de memoria columnar de Arrow permite análisis de alto rendimiento, interoperabilidad sin copia con pandas, Polar y DuckDB, y una eficiente E/S de archivos Parquet sin crear objetos en Python fila por fila.

El mssql-python controlador no requiere ninguna dependencia externa en máquinas Windows. El controlador instala todo lo que necesita con una sola pip instalación, así que puedes usar la última versión del controlador para nuevos scripts sin romper otros scripts que no tienes tiempo de actualizar y probar.

Documentación de mssql-python | Código fuente de mssql-python | Paquete (PyPI) | uv

Prerequisites

• Python 3.10 o posterior

• Si aún no tiene Python, instale el entorno de ejecución de Python y el administrador de paquetes pip desde python.org.

• ¿No desea usar su propio entorno? Sigue las instrucciones de Contenedores y desarrollo local para crear un devcontainer reproducible o un entorno de GitHub Codespaces.

• Visual Studio Code con las siguientes extensiones:

  • Python extensión para Visual Studio Code

• Interfaz de Azure Command-Line (CLI) para la autenticación sin contraseña en macOS y Linux.

• Si aún no tiene uv, siga las instrucciones de instalación.

• Una base de datos en SQL Server, Azure SQL Database o SQL Database en Fabric con el AdventureWorks2025 esquema de ejemplo y una cadena de conexión válida.

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

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

Creación de una base de datos SQL

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

Creación del proyecto y ejecución del código

1. Creación de un nuevo proyecto
2. Agregar dependencias
3. Inicio de Visual Studio Code
4. Actualizar pyproject.toml
5. Actualizar main.py
6. Guardar la cadena de conexión
7. Usa uv run para ejecutar el script

Creación de un nuevo proyecto

1. Abra una ventana del terminal en el directorio de desarrollo. Si no tienes uno, crea un nuevo directorio, como python o scripts. Evita las carpetas en tu OneDrive, ya que la sincronización puede interferir en la gestión de tu entorno virtual.

2. Crea un nuevo proyecto usando uv.

   uv init arrow-qs
   cd arrow-qs
   

Agregar dependencias

En el mismo directorio, instala los mssql-pythonpaquetes, python-dotenv, pyarrow, y rich .

uv add mssql-python python-dotenv pyarrow rich

Iniciar Visual Studio Code

En el mismo directorio, ejecute el siguiente comando.

code .

Actualizar pyproject.toml

1. El archivo pyproject.toml contiene los metadatos de tu proyecto. Abra el archivo en su editor favorito.

2. Revise el contenido del archivo. Debe ser similar a este ejemplo. Ten en cuenta la versión de Python y las dependencias para mssql-python; usa >= para definir una versión mínima. Si prefiere una versión exacta, cambie el >= valor anterior al número de versión a ==. Las versiones resueltas de cada paquete se almacenan en uv.lock. El lockfile garantiza que los desarrolladores que trabajan en el proyecto utilicen versiones de paquetes consistentes. Confirma tanto pyproject.toml como uv.lock, y ejecuta un escáner de dependencias aprobado por la organización en CI. No edites el uv.lock archivo directamente.

   [project]
   name = "arrow-qs"
   version = "0.1.0"
   description = "Add your description here"
   readme = "README.md"
   requires-python = ">=3.11"
   dependencies = [
       "mssql-python>=1.5.0",
       "pyarrow>=19.0.0",
       "python-dotenv>=1.1.1",
       "rich>=14.1.0",
   ]
   

3. Actualice la descripción para que sea más descriptivo.

   description = "Fetch SQL Server data as Apache Arrow tables using mssql-python"
   

4. Guarde y cierre el archivo.

Actualizar main.py

1. Abra el archivo denominado main.py. Debe ser similar a este ejemplo.

   def main():
       print("Hello from arrow-qs!")
   
   if __name__ == "__main__":
       main()
   

2. Reemplace el contenido completo de main.py con el siguiente código.

   """Fetch SQL Server data as Apache Arrow tables using mssql-python."""
   
   from os import getenv
   
   import pyarrow as pa
   import pyarrow.parquet as pq
   from dotenv import load_dotenv
   from mssql_python import connect, Connection
   from rich.console import Console
   from rich.table import Table
   
   console = Console()
   
   
   def get_connection() -> Connection:
       """Create a connection using the connection string from .env."""
       load_dotenv()
       conn_str = getenv("SQL_CONNECTION_STRING")
       if not conn_str:
           raise ValueError("SQL_CONNECTION_STRING not set in .env file")
       return connect(conn_str)
   
   
   def fetch_arrow_table(conn: Connection) -> pa.Table:
       """Run a query and return the full result as an Arrow Table."""
       cursor = conn.cursor()
       cursor.execute("""
           SELECT
               p.ProductID,
               p.Name,
               p.ProductNumber,
               p.Color,
               p.StandardCost,
               p.ListPrice,
               p.Size,
               p.Weight,
               p.SellStartDate,
               pc.Name AS Category
           FROM SalesLT.Product AS p
           INNER JOIN SalesLT.ProductCategory AS pc
               ON p.ProductCategoryID = pc.ProductCategoryID
           ORDER BY p.ListPrice DESC
       """)
       arrow_table = cursor.arrow()
       cursor.close()
       return arrow_table
   
   
   def fetch_arrow_batches(conn: Connection) -> pa.Table:
       """Stream results one batch at a time using arrow_batch()."""
       cursor = conn.cursor()
       cursor.execute("""
           SELECT
               c.CustomerID,
               c.CompanyName,
               c.EmailAddress,
               COUNT(soh.SalesOrderID) AS OrderCount,
               SUM(soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalSpent
           FROM SalesLT.Customer AS c
           LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh
               ON c.CustomerID = soh.CustomerID
           GROUP BY
               c.CustomerID,
               c.CompanyName,
               c.EmailAddress
           ORDER BY TotalSpent DESC
       """)
       batches = []
       while True:
           batch = cursor.arrow_batch()
           if batch is None or batch.num_rows == 0:
               break
           batches.append(batch)
       cursor.close()
   
       if not batches:
           return pa.table({})
   
       return pa.Table.from_batches(batches)
   
   
   def fetch_with_reader(conn: Connection) -> pa.Table:
       """Use arrow_reader() to stream results as a RecordBatchReader."""
       cursor = conn.cursor()
       cursor.execute("""
           SELECT
               soh.SalesOrderID,
               soh.OrderDate,
               (soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalDue,
               c.CompanyName
           FROM SalesLT.SalesOrderHeader AS soh
           INNER JOIN SalesLT.Customer AS c
               ON soh.CustomerID = c.CustomerID
           ORDER BY soh.OrderDate DESC
       """)
       reader = cursor.arrow_reader()
       arrow_table = reader.read_all()
       cursor.close()
       return arrow_table
   
   
   def display_arrow_table(arrow_table: pa.Table, title: str, max_rows: int = 10) -> None:
       """Display an Arrow table using rich formatting."""
       rich_table = Table(title=title)
   
       for name in arrow_table.column_names:
           rich_table.add_column(name, style="bright_white")
   
       for i in range(min(max_rows, arrow_table.num_rows)):
           row = [str(arrow_table.column(col)[i].as_py()) for col in range(arrow_table.num_columns)]
           rich_table.add_row(*row)
   
       if arrow_table.num_rows > max_rows:
           rich_table.add_row(*[f"... ({arrow_table.num_rows - max_rows} more rows)" if col == 0 else "" for col in range(arrow_table.num_columns)])
   
       console.print(rich_table)
       console.print(f"\n[dim]Schema: {arrow_table.num_columns} columns, {arrow_table.num_rows} rows[/dim]\n")
   
   
   def save_to_parquet(arrow_table: pa.Table, file_path: str) -> None:
       """Save an Arrow table to a Parquet file."""
       pq.write_table(arrow_table, file_path)
       console.print(f"[green]Saved {arrow_table.num_rows} rows to {file_path}[/green]\n")
   
   
   def main() -> None:
       conn = get_connection()
   
       # 1. Fetch entire result as an Arrow Table with cursor.arrow()
       console.rule("[bold]cursor.arrow() - Full table fetch[/bold]")
       products = fetch_arrow_table(conn)
       display_arrow_table(products, "Products (Top 10 by List Price)")
   
       # 2. Stream results in batches with cursor.arrow_batch()
       console.rule("[bold]cursor.arrow_batch() - Batch streaming[/bold]")
       customers = fetch_arrow_batches(conn)
       display_arrow_table(customers, "Customers by Total Spent")
   
       # 3. Use RecordBatchReader with cursor.arrow_reader()
       console.rule("[bold]cursor.arrow_reader() - RecordBatchReader[/bold]")
       orders = fetch_with_reader(conn)
       display_arrow_table(orders, "Recent Orders")
   
       # 4. Save to Parquet
       console.rule("[bold]Save to Parquet[/bold]")
       save_to_parquet(products, "products.parquet")
   
       # 5. Read back from Parquet and verify
       loaded = pq.read_table("products.parquet")
       console.print(f"[green]Read back {loaded.num_rows} rows from products.parquet[/green]")
       console.print(f"[dim]Schema: {loaded.schema}[/dim]\n")
   
       conn.close()
   
   
   if __name__ == "__main__":
       main()
   

Guardar la cadena de conexión

1. Abra el .gitignore archivo y agregue una exclusión para .env los archivos. El archivo debe ser similar a este ejemplo. Asegúrese de guardarlo y cerrarlo cuando haya terminado.

   # Python-generated files
   __pycache__/
   *.py[oc]
   build/
   dist/
   wheels/
   *.egg-info
   
   # Virtual environments
   .venv
   
   # Connection strings and secrets
   .env
   
   # Generated data files
   *.parquet
   

2. En el directorio actual, cree un nuevo archivo denominado .env.

3. En el .env archivo, agregue una entrada para la cadena de conexión denominada SQL_CONNECTION_STRING. Reemplace el ejemplo aquí por el valor real de la cadena de conexión.

   SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
   

   Importante

   Mantén .env en local y fuera del control de versiones. Para entornos de CI y entornos implementados, inyecta la cadena de conexión o los secretos de sus componentes desde el almacén de secretos de tu plataforma en lugar de copiar .env entre máquinas.

   Tip

   La cadena de conexión que usas depende en gran medida del tipo de base de datos SQL a la que te conectes. Si se conecta a una base de datos de Azure SQL o a una base de datos SQL en Fabric, use la cadena de conexión ODBC de la pestaña Cadenas de conexión. Es posible que tenga que ajustar el tipo de autenticación en función de su escenario. Para obtener más información sobre las cadenas de conexión y su sintaxis, consulte referencia de sintaxis de cadena de conexión.

Usa uv run para ejecutar el script

• En la ventana de terminal desde antes o una nueva ventana de terminal abierta en el mismo directorio, ejecute el siguiente comando.

  uv run main.py
  

  El script muestra tres métodos de obtención de Arrow:

  • cursor.arrow() devuelve un pyarrow.Table completo con todas las filas. Es ideal para conjuntos de resultados pequeños o medianos donde necesitas el conjunto de datos completo en memoria.

  • cursor.arrow_batch() devuelve pyarrow.RecordBatch de uno en uno. Lo ideal es para conjuntos de resultados grandes donde quieres procesar datos de forma incremental sin cargar todo en la memoria.

  • cursor.arrow_reader() devuelve un pyarrow.RecordBatchReader para transmisión en flujo. Más adecuado para el procesamiento en canalización o para pasarlo directamente a bibliotecas que aceptan un lector.

  El script también guarda los datos del producto en un archivo Parquet y los lee para verificar el viaje de ida y vuelta.

Funcionamiento del código

1. Conexión: El script carga la cadena de conexión desde un .env archivo y crea una conexión usando mssql_python.connect().

2. Obtención de tabla completa: cursor.arrow() ejecuta la consulta y devuelve todo el conjunto de resultados como un pyarrow.Tablearchivo . El controlador convierte los datos de su capa C++ usando la Interfaz de Datos Arrow C, evitando la creación de objetos en Python para mejorar el rendimiento.

3. Transmisión por lotes: cursor.arrow_batch() devuelve un pyarrow.RecordBatch por llamada. El bucle recoge lotes hasta que no quedan más filas, y luego los combina en una sola tabla. Utiliza este enfoque para grandes conjuntos de datos o cuando quieras procesar cada lote de forma independiente.

4. RecordBatchReader: cursor.arrow_reader() devuelve un pyarrow.RecordBatchReader, una interfaz Arrow estándar que muchas bibliotecas aceptan directamente. Llamar a reader.read_all() consume el flujo completo y lo convierte en una tabla.

5. E/S de Parquet: pyarrow.parquet.write_table() guarda la tabla Arrow en un archivo Parquet comprimido. Este formato preserva los tipos de columnas y soporta lecturas parciales eficientes.

Pasos siguientes

Utiliza estos artículos para seguir creciendo:

• Integración con Arrow para patrones avanzados de Arrow, incluidos el procesamiento por lotes, la gestión de memoria y la interoperabilidad con bibliotecas.
• integración con pandas para cargar los resultados de la consulta directamente en DataFrames.
• Integración con Polars para crear dataframes de Polars a partir de consultas nativas en Arrow.