クイックスタート: Python 用 mssql-python ドライバーで Apache Arrow を使用する

このクイックスタートでは、mssql-pythonドライバーの組み込みArrowフェッチメソッドを使って、SQL Serverデータをカラム形式のApache Arrowテーブルとして取得します。 Arrowの列状メモリ形式は、高性能な分析、pandas、Polars、DuckDBとのゼロコピーの相互運用性、そして行ごとのPythonオブジェクト作成なしで効率的なParquetファイルI/Oを可能にします。

mssql-python ドライバーでは、Windows マシンへの外部依存関係は必要ありません。ドライバーは必要なものをすべて1 pip インストールでインストールできるので、新しいスクリプトの最新バージョンを使い、アップグレードやテストに時間が取れない他のスクリプトを壊すことはありません。

mssql-python のドキュメント | mssql-python ソースコード | パッケージ (PyPI) | Uv

前提条件

Python 3.10 以降
Python をまだお持ちでない場合は、python.org から Python ランタイムと pip パッケージマネージャーをインストールします。
自分の環境を使いたくないですか? コンテナやローカル開発に従って、再現可能な開発コンテナやGitHub Codespaces環境を作成しましょう。
Visual Studio Code と次の拡張機能:
- Visual Studio Code 用の Python 拡張機能
macOS および Linux でのパスワードレス認証用の Azure Command-Line インターフェイス (CLI)。
uvがまだない場合は、インストール手順に従います。
Fabric の SQL Server、Azure SQL Database、または SQL データベースにある、AdventureWorks2025 サンプルスキーマと有効な接続文字列を備えたデータベース。

オペレーティングシステム固有の 1 回限りの前提条件をインストールします。 Windowsユーザーはこのステップをスキップできます。プラットフォームの詳細は「Install 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

SQL データベースを作成する

以下のいずれかのプラットフォームでSQLデータベースを作成または接続してください:

Azure portal を使用して数分で SQL データベースを作成する

新しいプロジェクトを作成する

開発ディレクトリでコマンドプロンプトを開きます。もし持っていなければ、 python や scriptsなどの新しいディレクトリを作成しましょう。 OneDriveのフォルダは避けてください。同期が仮想環境の管理に支障をきたす可能性があります。
を使って新しいuvを作成します。
```
uv init arrow-qs
cd arrow-qs
```

依存関係を追加する

同じディレクトリ内に mssql-python、 python-dotenv、 pyarrow、 rich パッケージをインストールします。

uv add mssql-python python-dotenv pyarrow rich

Visual Studio Code を起動します

同じディレクトリで、次のコマンドを実行します。

code .

pyproject.toml を更新する

pyproject.tomlファイルにはプロジェクトのメタデータが含まれています。お気に入りのエディターでファイルを開きます。
ファイルの内容を確認します。この例のようになります。最小バージョンを定義するためのPythonバージョンと依存関係mssql-python>=に注目してください。正確なバージョンを使用する場合は、バージョン番号の前の >= を ==に変更します。その後、各パッケージの解決済みバージョンが uv.lock に格納されます。ロックファイルにより、プロジェクトに携わる開発者が一貫したパッケージバージョンを使用することが保証されます。 pyproject.tomlとuv.lockの両方をコミットし、CIで組織承認の依存スキャナーを実行してください。 uv.lockファイルを直接編集しないでください。
```
[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",
]
```

説明をよりわかりやすいものに更新します。

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

ファイルを保存して閉じます。

main.py の更新

main.pyという名前のファイルを開きます。この例のようになります。
```
def main():
    print("Hello from arrow-qs!")

if __name__ == "__main__":
    main()
```

main.py のコンテンツ全体を次のコードに置き換えます。

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

接続文字列を保存する

.gitignore ファイルを開き、.env ファイルの除外を追加します。ファイルは次の例のようになります。完了したら、必ず保存して閉じてください。
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet
```
現在のディレクトリに、 .envという名前の新しいファイルを作成します。
.env ファイル内に、SQL_CONNECTION_STRINGという名前の接続文字列のエントリを追加します。ここでの例を実際の接続文字列の値に置き換えます。
```
SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
```
Important

.env はローカルに保持し、ソース管理の対象外にしてください。 CIやデプロイされた環境では、接続文字列やそのコンポーネントの秘密をプラットフォームの秘密ストアから注入し、マシン間で.envをコピーするのを防ぎましょう。

ヒント

使用する接続文字列は、接続するSQLデータベースの種類によって大きく異なります。 Fabric で Azure SQL Database または SQL データベースに接続する場合は、[接続文字列] タブから ODBC 接続文字列を使用します。シナリオによっては、認証の種類の調整が必要になる場合があります。接続文字列とその構文の詳細については、接続文字列の構文リファレンスを参照してください。

uv run を使用してスクリプトを実行する

ヒント

macOS では、 ActiveDirectoryInteractive と ActiveDirectoryDefault の両方が Microsoft Entra 認証で機能します。 ActiveDirectoryInteractive スクリプトを実行するたびにサインインするように求められます。繰り返しサインインのプロンプトを避けるために、を実行してaz loginから一度サインインし、その後キャッシュされた認証情報を再利用するActiveDirectoryDefaultを使いましょう。

以前のターミナルウィンドウで、または同じディレクトリに対して新しいターミナルウィンドウを開き、次のコマンドを実行します。
```
uv run main.py
```
このスクリプトは3つのアローフェッチ方式を示しています:
- cursor.arrow() すべての行で完全な pyarrow.Table を返します。メモリに全データセットが必要な小規模から中規模の結果セットに最適です。
- cursor.arrow_batch() は、一度に 1 つの pyarrow.RecordBatch を返します。データを逐分処理し、すべてをメモリにロードせずに処理したい大規模な結果セットに最適です。
- cursor.arrow_reader() は、ストリーミング用の pyarrow.RecordBatchReader を返します。パイプライン形式の処理や、リーダーを受け入れるライブラリに直接渡すのに最適です。
スクリプトはまた、製品データをParquetファイルに保存し、往復の確認のために読み戻します。

コードのしくみ

接続:スクリプトは.envファイルから接続文字列を読み込み、mssql_python.connect()を使って接続を作成します。
フルテーブルフェッチ: cursor.arrow() クエリを実行し、結果セット全体を pyarrow.Tableとして返します。ドライバはArrow Cデータインターフェースを使ってC++層でデータを変換し、Pythonオブジェクト作成をバイパスしてパフォーマンスを向上させます。
バッチストリーミング: cursor.arrow_batch() 通話ごとに1 pyarrow.RecordBatch を返します。ループは行がなくなるまでバッチを集め、それらを1つのテーブルにまとめます。この方法は大規模なデータセットや各バッチを独立して処理したい場合に使います。
RecordBatchReader: cursor.arrow_reader()pyarrow.RecordBatchReaderを返します。これは多くのライブラリが直接受け入れている標準的なArrowインターフェースです。 reader.read_all()を呼び出すと、ストリーム全体をテーブルにまとめます。
Parquet I/O: pyarrow.parquet.write_table() Arrowテーブルを圧縮されたParquetファイルに保存します。この形式は列の種類を保持し、効率的な部分読み取りをサポートします。

次のステップ

これらの記事を使って、さらに成長を続けましょう:

バッチ処理、メモリ管理、ライブラリ相互運用性を含む高度なArrowパターンのためのArrow統合。
pandasの統合により、クエリ結果を直接DataFramesに読み込むことができます。
Polars統合により、ArrowネイティブクエリからPolarsのデータフレームを構築します。

矢印積分

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-07-22