Polarsでmssql-pythonを使いましょう

PolarsはRustで書かれた高性能なDataFrameライブラリで、パンダに代わる高速でメモリ効率の良い代替を提供します。 Polarsとmssql-pythonドライバーを組み合わせることで、以下が可能です:

• SQLクエリ結果を直接Polars DataFrameに読み込みます。
• Microsoft SQLからのゼロコピーデータ転送にはApache Arrowを使いましょう。
• Polarsのデータフレームを効率的にMicrosoft SQLに書き戻す。
• 怠惰な評価で高性能なデータパイプラインを構築しましょう。

この記事の例は AdventureWorks サンプルデータベースをクエリします。 もしまだ持っていなければ、 AdventureWorksのサンプルデータベースをご覧ください。

Polars DataFramesにデータを読み込みます

MicrosoftのSQLデータをPolarsに読み込む方法は2通りです:標準的なカーソル方式による行ごとの変換、またはApache Arrowによるゼロコピー転送です。 「ゼロコピー」とは、データが単一のメモリバッファに留まり、ドライバー、Arrow、Polarが直接読み込むため、行が中間のPythonオブジェクトに重複しないようにすることを意味します。 この効率性のために、ほとんどのワークロードにはArrowアプローチを用いてください。

DataFrameへの基本クエリ

この方法は標準カーソルで全行を取得し、Polarsデータフレームを手動で構築します。 安全な値置換のための パラメータ化されたクエリ を受け付けています。 PyArrowなしでも動作しますが、すべての値をPythonを通過するため、大きな結果セットでは遅くなります。

import polars as pl
import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()


def query_to_polars(cursor, query: str, params: dict = None) -> pl.DataFrame:
    """Execute query and return results as Polars DataFrame."""
    cursor.execute(query, params or {})

    columns = [col[0] for col in cursor.description]
    rows = cursor.fetchall()

    data = {col: [row[i] for row in rows] for i, col in enumerate(columns)}
    return pl.DataFrame(data)

# Usage: %(color)s is a parameterized placeholder. The driver safely substitutes
# the value from the dict, which prevents SQL injection.
df = query_to_polars(cursor, "SELECT TOP 5 Name, ListPrice FROM Production.Product WHERE Color = %(color)s", {"color": "Black"})
print(df)

ゼロコピー転送にはArrowを使います(推奨)

MicrosoftのSQLデータをPolarsに読み込む最も効率的な方法はApache Arrowを利用することです。 mssql-pythonドライバの arrow() メソッドは、Polarがコピーオーバーヘッドゼロで利用可能な pyarrow.Table を返します。

def query_to_polars_arrow(cursor, query: str, params: dict = None) -> pl.DataFrame:
    """Execute query and load results through Arrow for best performance."""
    cursor.execute(query, params or {})
    arrow_table = cursor.arrow()
    return pl.from_arrow(arrow_table)

# Usage
df = query_to_polars_arrow(cursor, "SELECT ProductID, Name, ListPrice FROM Production.Product")
print(df)

Arrowバッチで大規模なデータセットをストリーミング

メモリに収まらないデータセットについては、 arrow_reader() を使ってストリーミングバッチで処理してください。 各バッチはPolarが独立して消費できる pyarrow.RecordBatch なので、メモリ使用量は結果全体ではなく、合計 batch_size に比例します。

def process_large_query(cursor, query: str, params: dict = None, batch_size: int = 50000) -> pl.DataFrame:
    """Process large query results as streaming Arrow batches."""
    cursor.execute(query, params or {})
    reader = cursor.arrow_reader(batch_size=batch_size)

    results = []
    for batch in reader:
        chunk_df = pl.from_arrow(batch)
        # Process each chunk
        results.append(chunk_df)

    return pl.concat(results) if results else pl.DataFrame()

# Usage
df = process_large_query(cursor, "SELECT * FROM Production.TransactionHistory")

遅延実行にはLazyFramesを使います

Polars LazyFramesは、すぐに実行せずに一連の操作(フィルター、グループ化、ソート)を構築できます。 Polarsは実行前にチェーン全体を最適化するため、各ステップを個別に適用するよりも高速です。

def query_to_lazy(cursor, query: str, params: dict = None) -> pl.LazyFrame:
    """Execute query and return a Polars LazyFrame."""
    cursor.execute(query, params or {})
    arrow_table = cursor.arrow()
    return pl.from_arrow(arrow_table).lazy()

# Build a query plan without executing immediately
lf = query_to_lazy(cursor, "SELECT SalesOrderID, CustomerID, TotalDue, OrderDate FROM Sales.SalesOrderHeader")
result = (
    lf.filter(pl.col("TotalDue") > 100)
    .group_by("CustomerID")
    .agg([
        pl.col("TotalDue").sum().alias("TotalSpent"),
        pl.col("SalesOrderID").count().alias("OrderCount")
    ])
    .sort("TotalSpent", descending=True)
    .collect()  # Execute the optimized plan
)
print(result)

Polars DataFrames を Microsoft SQL に書き込む

SQLインジェクションを防ぐためのクォート識別子

SQLではテーブル名や列名をクエリパラメータとして渡すことはできません。 動的識別子を持つSQL文を作成する際は、各名前を括弧で囲み、埋め込み ] 文字は脱出してSQLインジェクションを防ぎましょう。

def quote_id(identifier: str) -> str:
    """Quote a Microsoft SQL identifier to prevent SQL injection.
    
    Wraps the name in square brackets and escapes any embedded ] characters.
    Raises ValueError if the identifier is empty or contains null bytes.
    """
    if not identifier or "\x00" in identifier:
        raise ValueError(f"Invalid identifier: {identifier!r}")
    escaped = identifier.replace("]", "]]")
    return f"[{escaped}]"

このセクションのヘルパー機能は、生成されたSQLのすべてのテーブル名および列名に対して quote_id() を使用します。

データフレーム行を挿入する

行ごとのアプローチは、 iter_rows(named=True) でデータフレームを反復し、1行あたり1 INSERT を実行します。 この方法はシンプルですが、大量の処理には各行がサーバーへの往復を必要とするため遅くなります。

def polars_to_sql(cursor, conn, df: pl.DataFrame, table: str) -> int:
    """Write Polars DataFrame to Microsoft SQL table."""
    columns = df.columns
    placeholders = ", ".join([f"%({col})s" for col in columns])
    col_list = ", ".join([quote_id(col) for col in columns])
    query = f"INSERT INTO {quote_id(table)} ({col_list}) VALUES ({placeholders})"

    rows_inserted = 0
    for row in df.iter_rows(named=True):
        params = {k: (None if v is None else v) for k, v in row.items()}
        cursor.execute(query, params)
        rows_inserted += 1

    conn.commit()
    return rows_inserted

# Usage
cursor.execute("CREATE TABLE #PolarsInsert (Name NVARCHAR(100), Price DECIMAL(10,2), CategoryID INT)")
df = pl.DataFrame({
    "Name": ["Product A", "Product B"],
    "Price": [29.99, 49.99],
    "CategoryID": [1, 2]
})
rows = polars_to_sql(cursor, conn, df, "#PolarsInsert")
print(f"Inserted {rows} rows")

一括挿入 (大規模なDataFrameに推奨)

大規模なデータフレームの場合は、ドライバーのbulkcopy()メソッドを使ってTDS(Tabular Data Stream)プロトコル(SQL Microsoft使うネイティブのワイヤープロトコル)上で行をまとめて送信します。 この方法は往復を最小限に抑え、行ごとの挿入よりも高速です。

def polars_to_sql_bulk(conn, df: pl.DataFrame, table: str) -> int:
    """Bulk insert Polars DataFrame using BCP for best performance."""
    rows = [tuple(None if v is None else v for v in row) for row in df.iter_rows()]

    cursor = conn.cursor()
    result = cursor.bulkcopy(table, rows)
    conn.commit()
    return result["rows_copied"]

# Usage
cursor.execute("CREATE TABLE ##PolarsBulk (Name NVARCHAR(50), Price FLOAT, CategoryID INT)")
conn.commit()
df = pl.DataFrame({
    "Name": ["Product A", "Product B", "Product C"],
    "Price": [29.99, 49.99, 19.99],
    "CategoryID": [1, 2, 1]
})
rows = polars_to_sql_bulk(conn, df, "##PolarsBulk")
print(f"Bulk inserted {rows} rows")

データ分析パターン

以下の例は、Microsoft SQLクエリとPolars変換を組み合わせた一般的な解析タスクを示しています。

集約クエリ

この例では、製品をサブカテゴリごとにグループ化し、SQLでカウントと価格の統計を計算し、その後Polars DataFrameにまとめを読み込みます。

def get_sales_summary(cursor) -> pl.DataFrame:
    """Get sales summary by subcategory."""
    cursor.execute("""
        SELECT
            sc.Name AS SubcategoryName,
            COUNT(*) AS ProductCount,
            AVG(p.ListPrice) AS AvgPrice,
            MIN(p.ListPrice) AS MinPrice,
            MAX(p.ListPrice) AS MaxPrice
        FROM Production.Product p
        JOIN Production.ProductSubcategory sc ON p.ProductSubcategoryID = sc.ProductSubcategoryID
        GROUP BY sc.Name
        ORDER BY ProductCount DESC
    """)
    return pl.from_arrow(cursor.arrow())

df = get_sales_summary(cursor)
print(df)

時系列分析

Microsoft SQLの時系列データを読み込み、Polars式を使ってローリング平均などの計算された列を追加します。

def get_daily_sales(cursor, start_date: str, end_date: str) -> pl.DataFrame:
    """Get daily sales and compute rolling statistics."""
    cursor.execute("""
        SELECT
            CAST(OrderDate AS DATE) AS Date,
            COUNT(*) AS OrderCount,
            SUM(TotalDue) AS Revenue
        FROM Sales.SalesOrderHeader
        WHERE OrderDate BETWEEN %(start)s AND %(end)s
        GROUP BY CAST(OrderDate AS DATE)
        ORDER BY Date
    """, {"start": start_date, "end": end_date})

    df = pl.from_arrow(cursor.arrow())

    # Add rolling 7-day average
    df = df.with_columns(
        pl.col("Revenue").rolling_mean(window_size=7).alias("RollingAvg")
    )
    return df

sales_df = get_daily_sales(cursor, "2013-01-01", "2013-12-31")
print(sales_df)

SQLデータをローカルファイルと結合する

PolarsのローカルCSVファイルと結合することで、MicrosoftのSQLデータを豊かにすることができます。 各ソースをデータフレームに読み込み、メモリでジョインします。

# Load SQL data via Arrow
cursor.execute("SELECT c.CustomerID, p.FirstName, p.LastName FROM Sales.Customer c JOIN Person.Person p ON c.PersonID = p.BusinessEntityID")
customers = pl.from_arrow(cursor.arrow())

# Load local CSV
orders = pl.read_csv("orders_export.csv")

# Join in Polars
result = customers.join(orders, on="CustomerID", how="inner")
print(result)

ETLパターン

Microsoft SQLクエリとPolars変換を組み合わせて、抽出、変換、ロード(ETL)パイプラインを構築します。 Polars式は変換ステップを、 bulkcopy() はロードを担当します。

抽出、変換、読み込み

この例では、Arrowを通じてアクティブな顧客データを抽出し、Polars式を用いたビジネスセグメンテーションロジックを適用し、結果を一括コピーで読み込みます。

def etl_pipeline(source_cursor, dest_conn):
    """ETL pipeline using Polars transformations."""

    # Extract: derive a per-customer summary from order history via Arrow
    source_cursor.execute("""
        SELECT
            CustomerID,
            COUNT(*) AS OrderCount,
            SUM(TotalDue) AS TotalSpent
        FROM Sales.SalesOrderHeader
        WHERE OrderDate > DATEADD(YEAR, -1, (SELECT MAX(OrderDate) FROM Sales.SalesOrderHeader))
        GROUP BY CustomerID
    """)
    df = pl.from_arrow(source_cursor.arrow())
    df = df.with_columns(pl.col("TotalSpent").cast(pl.Float64))

    # Transform with Polars expressions
    df = df.with_columns([
        pl.when(pl.col("TotalSpent") > 1000).then(pl.lit("Platinum"))
          .when(pl.col("TotalSpent") > 500).then(pl.lit("Gold"))
          .when(pl.col("TotalSpent") > 100).then(pl.lit("Silver"))
          .otherwise(pl.lit("Bronze"))
          .alias("CustomerSegment"),
        (pl.col("TotalSpent") / pl.col("OrderCount").clip(lower_bound=1))
          .alias("AvgOrderValue"),
        (pl.col("TotalSpent") > 500).alias("IsHighValue")
    ])

    # Load via bulk copy into the destination table
    dest_cursor = dest_conn.cursor()
    dest_cursor.execute("""
        CREATE TABLE ##CustomerAnalytics (
            CustomerID INT,
            CustomerSegment NVARCHAR(20),
            AvgOrderValue FLOAT,
            IsHighValue BIT
        )
    """)
    dest_conn.commit()

    load_df = df.select(["CustomerID", "CustomerSegment", "AvgOrderValue", "IsHighValue"])
    polars_to_sql_bulk(dest_conn, load_df, "##CustomerAnalytics")

    return len(df)

パフォーマンスに関するヒント

以下のヒントは、mssql-pythonとPolarsの組み合わせを最大限に活用するのに役立ちます。

重労働はMicrosoft SQLに任せましょう

Microsoft SQLは、生データを有線で引き寄せてローカルでPython処理するよりも、集約、フィルタリング、ジョインが速いです。 可能な限りMicrosoft SQLに重労働を任せ、必要なデータだけをネットワーク上で移動させ、解析や変換にはPolarsを使い、Pythonでより便利なようにしましょう。

# Avoid: pulling all rows over the wire to aggregate locally in Polars
df = query_to_polars_arrow(cursor, "SELECT * FROM Production.Product WHERE Color IS NOT NULL")  # transfers entire table
summary = df.group_by("Color").agg(pl.col("ListPrice").sum())  # aggregation that SQL can do faster

# Better: push the aggregation into SQL and transfer only the summary
df = query_to_polars_arrow(cursor, """
    SELECT Color AS Category, SUM(ListPrice) AS TotalAmount
    FROM Production.Product
    WHERE Color IS NOT NULL
    GROUP BY Color
""")

すべての読み取り操作にはArrowを使用してください

Arrowベースの転送は中間のPythonオブジェクトを作成することを避け、メモリ使用を削減しスループットを向上させます。 数行以上の結果セットについては、手動の行ごとの変換よりも cursor.arrow() を好む。

# Suboptimal: Row-by-row conversion
cursor.execute("SELECT * FROM Production.TransactionHistory")
columns = [col[0] for col in cursor.description]
rows = cursor.fetchall()
df = pl.DataFrame({col: [row[i] for row in rows] for i, col in enumerate(columns)})

# Better: Arrow-based transfer
cursor.execute("SELECT * FROM Production.TransactionHistory")
df = pl.from_arrow(cursor.arrow())

関連するコンテンツ

• Apache Arrow との統合
• パンダス統合
• 一括コピー作業
• データの取得
• パフォーマンスチューニング