mssql-pythonアプリケーション向けのパフォーマンスチューニング

mssql-pythonドライバーは、接続プーリング、クエリ最適化、バルク操作など、SQL Serverアプリケーションのパフォーマンス最適化のための複数の機能とパターンを提供します。

接続管理

接続プーリングを使用する

接続プーリングは内蔵されています。 conn.close()を呼び出すと、接続は破壊されるのではなくプールに戻って再利用されるため、その後のconnect()通話では高価なハンドシェイクをスキップします。

import mssql_python

def get_data():
    conn = mssql_python.connect(
        "Server=<server>.database.windows.net;Database=<database>;"
        "Authentication=ActiveDirectoryDefault;Encrypt=yes"
    )
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
        return cursor.fetchall()
    finally:
        conn.close()

ワークロードに応じたプールサイズを設定する

並行処理の要件に応じてプールのサイズを調整してください。もしアプリケーションが複数の同時ユーザーを処理しているなら、プールを増やしましょう。軽負荷の場合は、より小さなプールがサーバーリソースを節約します:

import mssql_python

mssql_python.pooling(
    max_size=50,      # Default is 100; reduce or increase for your workload
    idle_timeout=600  # Seconds before idle connections are recycled
)

オペレーション内で接続を再利用

クエリごとに新しい接続を開くと、プーリングがあってもオーバーヘッドが増えます。代わりに、論理操作の間、単一の接続を保持します:

# Bad: New connection per query
def bad_pattern(product_ids):
    for pid in product_ids:
        conn = mssql_python.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
        row = cursor.fetchone()
        process(row)
        conn.close()

# Good: Single connection for all queries
def good_pattern(product_ids):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    try:
        for pid in product_ids:
            cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
            row = cursor.fetchone()
            process(row)
    finally:
        conn.close()

長期運行のサービスでは接続を開けておく

ウェブサーバー、キューワーカー、スケジュールされたジョブが連続して実行される場合は、接続を開いたままにしておくべきで、毎回接続・切断を行うのではなく、接続を開いたままにすべきです。接続の開通にはTCPハンドシェイク、TLS交渉、認証が必要で、ネットワーク距離や認証方法によって50〜200msかかることがあります。 1時間に何千ものメッセージを処理するキューワーカーにとっては、そのオーバーヘッドはすぐに蓄積されます。

作業員の生涯中は接続を開いたままにし、接続が切れたら再接続してください。キューが空のときにサーバーに負荷をかけないように、反復間でスリープをしましょう:

import mssql_python
import time

def run_worker(connection_string: str, poll_interval: float = 1.0):
    conn = None
    try:
        while True:
            try:
                if conn is None:
                    conn = mssql_python.connect(connection_string)
                cursor = conn.cursor()
                cursor.execute("SELECT TOP 1 * FROM dbo.JobQueue WHERE Status = 'Pending' ORDER BY CreatedDate")
                job = cursor.fetchone()
                if job:
                    try:
                        process_job(job)
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Done' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    except Exception:
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Failed' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    conn.commit()
                else:
                    time.sleep(poll_interval)  # No work available, wait before polling again
            except mssql_python.OperationalError:
                # Connection lost, reconnect on next iteration
                conn = None
                time.sleep(poll_interval)
    finally:
        if conn is not None:
            conn.close()

接続プーリングがデフォルトで有効になっている場合、プールがアイドル接続を処理してくれます。しかし、プーリングを無効にしたり専用接続を1つ使う場合は、接続文字列でConnection TimeoutとCommand Timeoutを設定して、古い接続を早期に検出し、ハングしないようにしましょう。

クエリの最適化

フェッチはデータのみを必要としていました

アプリケーションが使用する列だけを選択することで、ネットワーク転送、メモリ消費、クエリ実行時間を短縮できます。

# Bad: Select all columns
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})

# Good: Select specific columns
cursor.execute("""
    SELECT SalesOrderID, OrderDate, TotalDue
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(customer_id)s
""", {"customer_id": 1})

適切な取得方法を使用する

ドライバは複数のフェッチメソッドを提供します。結果のサイズに合うものを使いましょう:

fetchval() 最小限のオーバーヘッドで単一のスカラー値を返します。
fetchall() 結果セット全体をメモリに読み込み、小さなテーブルには適しています。
fetchmany(n) 行をバッチ単位で取得し、大きな結果セットに対してメモリ使用量を一定に保ちます。

fetchmany()に適したバッチサイズは行幅によります。狭い行(小さな列が数行、それぞれ約1KB)の場合、1,000行で各バッチのメモリは約1MB程度に保たれます。大きな文字列や二進列の幅広の行には、より小さなバッチサイズを使いましょう。まず1,000から始めて、データに基づいて調整してください。

def process_batch(rows):
    # Example: print each row. Replace with your own logic.
    for row in rows:
        print(row)

# Single value
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()

# Small result set
cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor.fetchall()

# Large result set - process in batches
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader")
while True:
    batch = cursor.fetchmany(1000)
    if not batch:
        break
    process_batch(batch)

サーバー側ページネーションを使います

すべての行を取得してPythonをスライスする代わりに、OFFSET/FETCH NEXTで必要なページだけを取得しましょう。

def get_page(cursor, page: int, page_size: int = 50) -> list:
    """Get paginated results efficiently."""
    offset = (page - 1) * page_size

    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY ProductID
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})

    return cursor.fetchall()

SET NOCOUNT ON を使用します

デフォルトでは、SQL ServerはDML文のたびに「影響を受けた行」メッセージを送信します。 SET NOCOUNT ON これらのメッセージを抑制し、ネットワークトラフィックを減少させます。セッションレベルの設定なので、すべてのクエリに埋め込むのではなく、接続後に一度だけ設定すればいいです。

# Set once after connecting
cursor.execute("SET NOCOUNT ON")

# All subsequent statements on this connection skip the row-count message
cursor.execute(
    "INSERT INTO Log (Message) VALUES (%(message)s)",
    {"message": "Log entry"}
)

適切な挿入方法を選んでください

ドライバーは異なるスケールに適した3つのデータ挿入方法を提供しています:

Method	行数	なぜでしょうか
`execute()`	通話ごとに1行	フォームの提出やAPIハンドラのように、挿入したIDをすぐに必要とする場合に単一行の操作に使うのが良いでしょう。
`executemany()`	~10〜1,000行	ループよりも高いスループットを実現するためにカラムごとのパラメータバインディングを使用します。各行をパラメータ化された文として送信します。
`bulkcopy()`	数百行以上	TDSの大量挿入プロトコルを使用しており、行ごとの挿入よりもはるかに効率的です。データの読み込み、移行、バッチ処理に最適です。

詳細や例については、データの読み込みと移動パターンを参照してください。

execute() を使用した単一挿入

すぐに結果が必要な場合は一度きりの挿入物に使ってください。 Production.Product デフォルトのない複数のNOT NULL列があるため、挿入時にそれらすべてが一覧化されます:

from datetime import datetime

cursor.execute(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (%(name)s, %(number)s, %(safety)s, %(reorder)s,
            %(cost)s, %(price)s, %(days)s, %(start)s)
    """,
    {
        "name": "Widget", "number": "WG-1001",
        "safety": 100, "reorder": 75,
        "cost": 12.50, "price": 19.99,
        "days": 1, "start": datetime(2024, 1, 1),
    },
)
conn.commit()

executemany() を使用したバッチ挿入

executemany() パラメータを列ごとにバインドし、効率的に送信します。ループで execute() を呼び出すのではなく、中規模のバッチ処理に使用してください。 executemany()はタプルリスト付きの位置?マーカーを必要とするのに対し、execute()はdict付きで、?パラメータと名前付き%(name)sパラメータの両方をサポートしています。各スタイルの詳細についてはパラメータ付きクエリを参照してください。

rows = [
    ("Widget A", "WG-1001", 100, 75, 12.50, 19.99, 1, datetime(2024, 1, 1)),
    ("Widget B", "WG-1002", 100, 75, 15.00, 24.99, 1, datetime(2024, 1, 1)),
    ("Widget C", "WG-1003", 100, 75, 18.00, 29.99, 1, datetime(2024, 1, 1)),
]

cursor.executemany(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """,
    rows,
)
conn.commit()

大量の荷物のための一括コピー

スループットが行単位の制御よりも重要であれば、 bulkcopy()に切り替えましょう。 TDSのバルクインサートプロトコルを通じて行をストリーミングし、パラメータ化された文の行ごとのオーバーヘッドを回避します。 bulkcopy() が executemany() を上回る正確なクロスオーバーポイントは、行幅やネットワーク遅延によって異なりますが、通常は100行台前半です。非常に小さなバッチの場合、 executemany() は bulkcopy() 別の内部接続を作成し自動的にコミットできるためよりシンプルです。

execute()やexecutemany()とは異なり、bulkcopy()はINSERT列リストではなく、位置ごとに値を列にマッピングします。 column_mappings を渡して読み込み先の列を指定すると、ソースタプルがテーブル先頭の IDENTITY 列ではなく、正しい列に対応するようになります:

result = cursor.bulkcopy(
    "Production.Product",
    rows,
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
)
print(f"Copied {result['rows_copied']} rows")

非常に大きな負荷の場合は、データセット全体をメモリにロードしないようにジェネレーターを使用し、 batch_size を定期的にコミットするように設定してください:

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy(
    "Production.Product",
    csv_rows("products.csv"),
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
    batch_size=5000,
)

キャッシュ戦略

ほとんど変わらない参照データ(カテゴリ、ルックアップテーブル、設定)については、すべてのリクエストにクエリするのではなく、アプリケーション内で結果をキャッシュしてください。

Pythonのfunctools.lru_cacheは簡単なメモ化を提供しますが、プロセスが再起動されるまで無期限にキャッシュされます。基礎データが変更される可能性がある場合は、制限時間後に自動的にリフレッシュする cachetools.TTLCache を使います。

from cachetools import TTLCache, cached

category_cache = TTLCache(maxsize=1, ttl=300)  # Refresh every 5 minutes

@cached(category_cache)
def get_categories(connection_string: str) -> tuple:
    conn = mssql_python.connect(connection_string)
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
        return cursor.fetchall()
    finally:
        conn.close()

ネットワークの最適化

往復を最小限に抑える

各クエリはサーバーへのネットワーク往復です。関連するクエリを一つのバッチにまとめ、 nextset() を使って結果セットを進めます:

# Bad: Multiple round trips
cursor.execute("SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
customer = cursor.fetchone()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
orders = cursor.fetchall()
cursor.execute("SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s)", {"customer_id": 1})
detail_count = cursor.fetchval()

# Good: Single round trip
cursor.execute("""
    SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s;
    SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s;
    SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s);
""", {"customer_id": 1})

customer = cursor.fetchone()
cursor.nextset()
orders = cursor.fetchall()
cursor.nextset()
detail_count = cursor.fetchval()

複雑な論理にはサーバーサイド処理を活用してください

生の行を取得してPythonで処理する代わりに、集約やフィルタリングをSQL Serverにプッシュしましょう。サーバーは数千行の詳細行ではなく、単一の要約行を返します:

cursor.execute("""
    SELECT p.Name, COUNT(sod.SalesOrderDetailID) AS OrderCount, SUM(sod.LineTotal) AS TotalSales
    FROM Production.Product p
    JOIN Sales.SalesOrderDetail sod ON p.ProductID = sod.ProductID
    WHERE p.ProductID = %(product_id)s
    GROUP BY p.Name
""", {"product_id": 707})

交互のカーソル操作を避ける

mssql-pythonドライバーは複数のアクティブ結果セット(MARS)をサポートしていません。接続ごとにアクティブなクエリを持つカーソルは1つだけです。次のクエリを実行する前に最初の結果セットを完全に取得するか、または2つ目の接続を使用します。

connection_string = (
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

# Option 1: Fetch first, then query (single connection)
# Warning: This is an N+1 pattern. Each iteration is a round trip.
# Use this only when the JOIN in Option 2 is not possible.
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT ProductID FROM Production.Product WHERE ProductSubcategoryID = 1")
product_ids = [row[0] for row in cursor.fetchall()]

for pid in product_ids:
    cursor.execute("SELECT ProductID, LocationID, Quantity FROM Production.ProductInventory WHERE ProductID = %(product_id)s", {"product_id": pid})
    inventory = cursor.fetchone()

conn.close()

# Option 2: Use a JOIN instead of N+1 queries (preferred)
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
    SELECT p.ProductID, p.Name, i.Quantity
    FROM Production.Product p
    LEFT JOIN Production.ProductInventory i ON p.ProductID = i.ProductID
    WHERE p.ProductSubcategoryID = 1
""")
results = cursor.fetchall()
conn.close()

メモリ管理

大量の結果をチャンク単位で処理する

数百万行のテーブルをリストに読み込むと、結果セット全体に比例したメモリ消費が発生します。 OFFSETとFETCH NEXTを使ってサーバー側のデータページを進め、1つずつチャンクを処理します。

def quote_id(identifier: str) -> str:
    """Quote a possibly schema-qualified SQL identifier to prevent SQL injection."""
    return ".".join("[" + part.replace("]", "]]") + "]" for part in identifier.split("."))

def process_large_table(cursor, table: str, columns: list[str], key_column: str, processor, chunk_size: int = 10000):
    """Process large table without loading all data."""
    safe_table = quote_id(table)
    safe_key = quote_id(key_column)
    col_list = ", ".join(quote_id(c) for c in columns)
    cursor.execute(f"SELECT COUNT(*) FROM {safe_table}")
    total = cursor.fetchval()

    offset = 0
    while offset < total:
        cursor.execute(f"""
            SELECT {col_list} FROM {safe_table}
            ORDER BY {safe_key}
            OFFSET ? ROWS
            FETCH NEXT ? ROWS ONLY
        """, (offset, chunk_size))

        chunk = cursor.fetchall()
        processor(chunk)

        offset += chunk_size
        print(f"Processed {min(offset, total)}/{total}")

# key_column must be unique, otherwise rows can be duplicated or skipped across pages
process_large_table(
    cursor,
    "Production.TransactionHistory",
    ["TransactionID", "ProductID", "Quantity", "ActualCost"],
    "TransactionID",
    lambda chunk: None,  # replace with your row-processing logic
)

ストリーミングにはジェネレーターを使う

Pythonジェネレーターのラッピングfetchmany()テーブルサイズに関係なくメモリ使用量を一定に保ちます。呼び出し者は結果セット全体を読み込まずに行ごとに反復を行います。超大型ソースの場合は、 UNION ALL とテーブルを組み合わせて同じ方法で結果をストリーミングします。

def stream_query(cursor, query: str, params: dict = None, batch_size: int = 1000):
    cursor.execute(query, params or {})
    
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        for row in batch:
            yield row

# Union the live and archive transaction tables into one extra-large result set
query = """
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistory
    UNION ALL
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistoryArchive
"""

count = 0
for row in stream_query(cursor, query, batch_size=5000):
    count += 1
print(f"Streamed {count} rows")

資源を迅速に片付けましょう

未接続はサーバーリソースを占有し、接続プールを使い果たすことがあります。例外が発生してもクリーンアップを保証するためにコンテキストマネージャーを使いましょう。

from contextlib import contextmanager

@contextmanager
def database_connection(connection_string: str):
    conn = mssql_python.connect(connection_string)
    try:
        yield conn
    finally:
        conn.close()

with database_connection(connection_string) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
    data = cursor.fetchall()

メモリ使用量の監視

大規模な結果セット、長寿命キャッシュ、接続オブジェクトはすべてメモリを消費します。アプリケーションがサービスとして動作する場合、閉じていないカーソルや境界のないキャッシュによるメモリリークが、最終的にOSやコンテナランタイムによってプロセスが終了する原因となることがあります。

Pythonのtracemallocモジュールを使ってメモリをスナップショットし、最大の割り当てを見つけてください。

import tracemalloc

tracemalloc.start()

# ... run your workload ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
    print(stat)

予期せぬ記憶増加の一般的な原因には以下があります:

何百万行も返すクエリで fetchall() 呼び出す。代わりに fetchmany() か発電機を使いましょう。
maxsizeやTTLなしでクエリ結果をキャッシュする方法。キャッシュはプロセスが再起動されるまで成長します。
閉じないままループ内でカーソルを作成すること。各開いたカーソルは、その結果セットをメモリに保持します。

インデックスおよびクエリプランの最適化

サーバーサイドのクエリ性能を確認してください

SET STATISTICS TIME ONとSET STATISTICS IO ONを使って、サーバー上でクエリにどれくらいかかるか、どれだけのデータが読み取られているかを確認しましょう。論理リードが高い場合は、通常インデックスが欠落していることを示します。これらの文はSQL Server Management StudioまたはVisual Studio CodeのMSSQL拡張機能で実行し、出力はメッセージペインに表示されます。

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

次のような出力が表示されます。

Table 'Product'. Scan count 1, logical reads 3
SQL Server Execution Times: CPU time = 0 ms, elapsed time = 1 ms.

論理リードやテーブルスキャンが多い場合は、インデックスの追加を検討してください。

クエリヒントを戦術的な解決策として活用しましょう

クエリヒントはクエリオプティマイザーのインデックスを上書きし、戦略の選択を結合します。本番環境では、クエリが突然後退した際の迅速かつ低リスクのパッチとして価値があります。根本原因(インデックスの欠落、古い統計、スキーマの変更など)を調査しながら、アプリケーションコードにヒントをすぐに導入してクエリを安定させることができます。

ヒントを永久に残しておくのは避けましょう。データ分布やスキーマが変わると、ハードコーディングされたヒントが状況を悪化させることがあります。一時的なものとして扱い、根本的な問題が解決した後に再検討してください:

cursor.execute("""
    SELECT * FROM Production.Product WITH (INDEX(AK_Product_Name))
    WHERE ProductSubcategoryID = %(subcategory_id)s
""", {"subcategory_id": 1})

OPTION(再コンパイル)を使って、悪いキャッシュプランを回避してください

SQL Serverは、最初に見たパラメータ値のセットに基づいてクエリ計画をキャッシュします。呼び出しごとにデータ分布が大きく異なる場合、キャッシュプランは一部の値で性能が悪くなることがあります。この問題はパラメータスニッフィングと呼ばれ、「かつては速かった」クエリが突然数秒や数分かかる形で現れることが多いです。

OPTION (RECOMPILE)SQL Serverは各実行ごとに新しい計画を立てることを強制し、サーバー側の変更なしで即座にデプロイできる効果的な修正となります。トレードオフとして1コールあたりのコンパイルコストは小さいですが、実行頻度が低いクエリや可変サイズの結果セットを返すクエリでは、悪いプランを実行する場合と比べればそのコストは無視できるほどです。

問題が安定したら、クエリの書き直し、フィルタリングされたインデックスの追加、プランガイドの活用など、恒久的な修正に時間をかけて進めることができます。

cursor.execute("""
    SELECT * FROM Sales.SalesOrderHeader
    WHERE OrderDate > %(start_date)s
    OPTION (RECOMPILE)
""", {"start_date": start_date})

パフォーマンスの監視

質問のタイミングを計ってください

遅い操作を見つけるには、クエリを time.perf_counter()でラップします:

import time

start = time.perf_counter()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": customer_id})
rows = cursor.fetchall()
elapsed = time.perf_counter() - start

print(f"Query returned {len(rows)} rows in {elapsed:.3f}s")

アプリケーションがどこに時間を費やしているかを広く把握するには、Pythonの内蔵のcProfileモジュールを活用してください。

python -m cProfile -s cumtime my_app.py

このビューは関数呼び出しごとの累積時間を示し、遅延がクエリ実行、データ処理、ネットワーク遅延のいずれかにあるかを識別するのに役立ちます。

サーバー側の解析にはクエリストアをご利用ください

クライアント側のタイミングは、アプリケーションの視点からクエリにかかる時間を示しますが、ネットワークの遅延、サーバー実行時間、クライアント処理を組み合わせたものです。クエリストアはサーバー上の実行計画やランタイム統計をキャプチャし、SQL Serverがどのようにクエリを実行し、どのくらいの頻度で実行され、パフォーマンスが時間とともにどのように変化したかを正確に把握できます。

クエリストアは、パラメータスニッフィング、プラン回帰、サーバーリソースを最も多く消費するクエリを特定するのに特に有用です。 sys.query_store_runtime_statsやsys.query_store_planビューを直接クエリすることも、SQL Server Management Studioの組み込みクエリストアレポートを使うこともできます。

パフォーマンスダッシュボードレポートの使用

SQL Server Management Studioのパフォーマンスダッシュボードレポートは、現在の待機タイプ、アクティブで高価なクエリ、CPU/IOの傾向など、SQL Serverの健康状態をリアルタイムで概観できます。 DMVに直接問い合わせを書くことなく、ボトルネックを素早く見つけられます。

パフォーマンスチェックリスト

Connection

[ ] 接続プーリングを有効にしてください。
[ ] 自分の仕事量に合ったプールのサイズを決めてください。
[ ] オペレーション内で接続を再利用してください。
[ ] 長期的なサービスで繋がりを保ち続けてください。

Queries

[ ] 必要な列だけを選択してください。
[ ] 各クエリごとに適切なフェッチ方法を使いましょう。
[ ] サーバーサイドのページ設定を実装してください。
[ ] 接続後に SET NOCOUNT ON を一度設定します。
[ ] クエリをバッチ処理して往復を最小限に抑えます。

挿入

[ ] execute() は単列インサートに使ってください。
[ ] 小〜中ロット(~10〜1,000段)には executemany() を使います。
[ ] スループットが行ごとの制御よりも重要な場合に bulkcopy() 使ってください。

キャッシュ処理

[ ] TTLで参照データをキャッシュして、古くなった結果を提供しないようにしましょう。

Resources

[ ] 大きな結果をチャンク単位またはジェネレーターで処理します。
[ ] 接続を迅速に片付けろ。
[ ] 長時間稼働するサービスでは tracemalloc でメモリ使用を監視してください。

フィードバック

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

Last updated on 2026-07-22