Performance Tuning für mssql-python-Anwendungen

Der mssql-python-Treiber bietet mehrere Funktionen und Muster zur Optimierung der SQL Server-Anwendungsleistung, darunter Connection Pooling, Abfrageoptimierung und Massenoperationen.

Verbindungsverwaltung

Verbindungspooling verwenden

Verbindungspooling ist integriert. Wenn Sie anrufen conn.close(), kehrt die Verbindung zum Pool zurück, um wiederverwendet zu werden, anstatt zerstört zu werden, sodass nachfolgende connect() Anrufe den teuren Handschlag überspringen:

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

Poolgröße für Workload konfigurieren

Passe die Poolgröße je nach deinen Anforderungen an die Konkurrenz an. Wenn Ihre Anwendung viele gleichzeitige Nutzer verarbeitet, erhöhen Sie den Pool. Bei leichteren Arbeitslasten spart ein kleinerer Pool Serverressourcen:

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
)

Wiederverwendung von Verbindungen innerhalb von Operationen

Das Öffnen einer neuen Verbindung für jede Abfrage erhöht den Overhead, selbst mit Pooling. Stattdessen halten Sie eine einzelne Verbindung für die Dauer einer logischen Operation:

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

Halten Sie Verbindungen in langjährigen Diensten offen

Webserver, Queue-Worker und geplante Aufgaben, die kontinuierlich ausgeführt werden, sollten Verbindungen offen halten, anstatt für jeden Vorgang erneut eine Verbindung herzustellen und wieder zu trennen. Das Öffnen einer Verbindung erfordert einen TCP-Handshake, TLS-Verhandlung und Authentifizierung, was je nach Netzwerkentfernung und Authentifizierungsmethode 50–200 ms dauern kann. Für einen Warteschlangenarbeiter, der tausende Nachrichten pro Stunde verarbeitet, summiert sich dieser Aufwand schnell.

Halten Sie die Verbindung für die gesamte Lebensdauer des Arbeiters offen und verbinden Sie sich erneut, wenn die Verbindung ausfällt. Warte zwischen den Iterationen, um den Server nicht zu überlasten, wenn die Warteschlange leer ist:

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

Mit aktiviertem Connection Pooling (dem Standard) übernimmt der Pool die Idle-Verbindungen für dich. Aber wenn du das Pooling deaktivierst oder eine einzelne dedizierte Verbindung verwendest, setze Connection Timeout und Command Timeout in deinem Verbindungszeichenfolge so, dass veraltete Verbindungen frühzeitig erkannt werden, anstatt dass sie hängen.

Abfrageoptimierung

Nur die benötigten Daten abrufen

Die Auswahl nur der Spalten, die Ihre Anwendung verwendet, reduziert den Netzwerktransfer, den Speicherverbrauch und die Abfrageausführungszeit.

# 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})

Verwenden Sie geeignete Fetch-Methoden

Der Treiber bietet mehrere Abrufmethoden an. Verwenden Sie die, die Ihrer Ergebnisgröße entspricht:

• fetchval() gibt einen einzelnen skalaren Wert mit minimalem Overhead zurück.
• fetchall() lädt die gesamte Ergebnismenge in den Speicher, was für kleine Tabellen gut funktioniert.
• fetchmany(n) ruft Zeilen in Chargen ab und hält die Speichernutzung für große Ergebnismengen konstant.

Die richtige Chargengröße für fetchmany() hängt von der Zeilenbreite ab. Für schmale Zeilen (einige kleine Spalten, jeweils etwa 1 KB) sorgen 1.000 Zeilen dafür, dass jeder Batch rund 1 MB Speicherbedarf hat. Für breitere Zeilen mit großen Strings oder binären Spalten verwenden Sie eine kleinere Chargengröße. Fang mit 1.000 an und passe dich basierend auf deinen Daten an.

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)

Verwenden Sie serverseitige Paginierung

Anstatt alle Zeilen zu holen und in Python zu slicen, nutze OFFSET/FETCH NEXT es, um nur die Seite abzurufen, die du brauchst.

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

Verwenden Sie SET NOCOUNT ON

Standardmäßig sendet SQL Server nach jeder DML-Anweisung eine Nachricht "Zeilen betroffen". SET NOCOUNT ON Unterdrückt diese Nachrichten und reduziert den Netzwerkverkehr. Es handelt sich um eine Session-Level-Einstellung, also setze sie einmal nach der Verbindung ein, anstatt sie in jede Abfrage einzubetten.

# 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"}
)

Wählen Sie die richtige Einfügungsmethode

Der Treiber bietet drei Möglichkeiten, Daten einzufügen, die jeweils auf eine andere Skala zugeschnitten sind:

Für weitere Details und Beispiele siehe Datenlade- und Bewegungsmuster.

Einzelne Einfügevorgänge mit execute()

Nutze sie für einmalige Einsätze, bei denen du das Ergebnis sofort brauchst. Production.Product hat mehrere Spalten mit NOT-NULL-Constraint ohne Standardwerte, daher listet die INSERT-Anweisung sie alle auf:

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

Batch-Einfügungen mit Executemany()

executemany() bindet Parameter spaltenweise und sendet sie effizient. Verwende es für mittelgroße Stapel, anstatt execute() in einer Schleife aufzurufen. Beachten Sie, dass executemany() positionale ?-Marker in Verbindung mit einer Liste von Tupeln erfordert, während execute() sowohl ? als auch benannte %(name)s-Parameter mit Dictionaries unterstützt. Siehe Parametrisierte Abfragen für Details zu jedem Stil.

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

Massenkopieren für große Datenmengen

Wenn der Durchsatz wichtiger ist als die Kontrolle pro Zeile, schalten Sie auf bulkcopy(). Es überträgt Zeilen über das TDS-Bulk-Insert-Protokoll und vermeidet den mit jeder Zeile verbundenen Overhead parametrisierter Anweisungen. Der genaue Punkt, ab dem bulkcopy() besser abschneidet als executemany(), hängt von der Zeilenbreite und der Netzwerklatenz ab, liegt aber typischerweise im niedrigen Hunderterbereich an Zeilen. Für sehr kleine Chargen ist executemany() einfacher, da bulkcopy() eine separate interne Verbindung erstellt und automatisch ein Commit ausführt.

Im Gegensatz zu execute() und executemany() ordnet bulkcopy() Werte Spalten anhand ihrer Position zu, nicht anhand einer INSERT-Spaltenliste. Gib column_mappings an, um die Zielspalten zu benennen, in die du lädst, damit die Quelltupel den richtigen Spalten zugeordnet werden statt der führenden IDENTITY-Spalte der Tabelle.

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

Für sehr große Datenmengen verwenden Sie einen Generator, um das Laden des gesamten Datensatzes im Speicher zu vermeiden, und setzen Sie batch_size so, dass regelmäßig committet wird:

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,
)

Caching-Strategien

Für Referenzdaten, die sich selten ändern (Kategorien, Lookup-Tabellen, Konfiguration), cachen Sie die Ergebnisse in Ihrer Anwendung, anstatt bei jeder Anfrage abzufragen.

Python functools.lru_cache bietet eine einfache Memoisierung, wird aber unbegrenzt gecachet, bis der Prozess neu gestartet wird. Wenn sich die zugrunde liegenden Daten ändern können, verwenden Sie cachetools.TTLCache, um nach Ablauf eines Zeitlimits automatisch zu aktualisieren:

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

Netzwerkoptimierung

Hin- und Rückfahrten minimieren

Jede Abfrage ist eine Netzwerk-Rundreise zum Server. Kombinieren Sie verwandte Abfragen zu einem einzigen Batch und verwenden Sie nextset(), um zu den Ergebnismengen weiterzuschalten:

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

Verwenden Sie serverseitige Verarbeitung für komplexe Logik

Verlagern Sie Aggregationen und Filterungen auf SQL Server, anstatt Rohdatenzeilen abzurufen und sie in Python zu verarbeiten. Der Server liefert eine einzelne Zusammenfassungszeile anstelle potenziell Tausender von Detailzeilen:

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})

Vermeiden Sie verschachtelte Cursor-Operationen

Der mssql-python-Treiber unterstützt keine Multiple Active Result Sets (MARS). Pro Verbindung kann nur ein Cursor eine aktive Abfrage haben. Holen Sie die erste Ergebnismenge vollständig ab, bevor Sie die nächste Abfrage ausführen, oder verwenden Sie eine zweite Verbindung:

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

Speicherverwaltung

Verarbeiten Sie große Ergebnisse in Blöcken

Das Laden einer Tabelle mit mehreren Millionen Zeilen in eine Liste verbraucht Speicher proportional zum vollständigen Ergebnissatz. Verwenden Sie OFFSET und FETCH NEXT, um die Daten serverseitig zu paginieren und jeweils einen Datenblock zu verarbeiten.

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
)

Verwenden Sie Generatoren zum Streaming

Ein Python-Generator-Wrapping fetchmany() hält den Speicherverbrauch unabhängig von der Tabellengröße konstant. Der Aufrufer iteriert Zeile für Zeile, ohne die vollständige Ergebnismenge zu laden. Für eine extra-große Quelle kombinieren Sie Tabellen mit UNION ALL und streamen Sie das kombinierte Ergebnis auf die gleiche Weise.

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

Räumen Sie Ressourcen umgehend auf

Ungeschlossene Verbindungen binden Serverressourcen und können den Verbindungspool erschöpfen. Verwenden Sie einen Kontextmanager, um eine Bereinigung zu gewährleisten, selbst wenn Ausnahmen auftreten.

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

Überwachen der Arbeitsspeicherauslastung

Große Ergebnismengen, langlebige Caches und Verbindungsobjekte verbrauchen alle Speicher. Wenn Ihre Anwendung als Dienst läuft, können Speicherlecks durch ungeschlossene Cursor oder unbegrenzte Caches schließlich dazu führen, dass der Prozess vom Betriebssystem oder der Containerlaufzeit gestoppt wird.

Verwenden Sie Pythons Modul tracemalloc, um eine Speicheraufnahme zu erstellen und die größten Speicherbelegungen zu finden.

import tracemalloc

tracemalloc.start()

# ... run your workload ...

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

Häufige Quellen für unerwartetes Gedächtniswachstum sind:

• Der Aufruf von fetchall() für eine Abfrage, die Millionen von Zeilen zurückgibt. Verwende stattdessen fetchmany() oder einen Generator.
• Zwischenspeichern von Abfrageergebnissen ohne maxsize oder TTL. Die Caches wachsen, bis der Prozess neu gestartet wird.
• Erstellen von Cursorn in einer Schleife, ohne sie zu schließen. Jeder offene Cursor speichert sein Ergebnis im Speicher.

Index- und Abfrageplanoptimierung

Überprüfen Sie die serverseitige Abfrageleistung

Nutze SET STATISTICS TIME ON und SET STATISTICS IO ON um zu sehen, wie lange Abfragen auf dem Server dauern und wie viele Daten sie lesen. Hohe logische Lesevorgänge weisen in der Regel auf einen fehlenden Index hin. Führen Sie diese Anweisungen im SQL Server Management Studio oder in der MSSQL-Erweiterung für Visual Studio Code aus, wobei die Ausgabe im Nachrichtenbereich angezeigt wird:

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

Die Ausgabe sollte wie folgt angezeigt werden:

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

Wenn Sie viele logische Lesevorgänge oder Tabellenscans feststellen, sollten Sie einen Index erstellen.

Nutze Anfragehinweise als taktische Lösung

Abfragehinweise überschreiben die Auswahl des Abfrageoptimierers für Index- und Join-Strategien. In der Produktion sind sie wertvoll als schneller, risikoarmer Patch, wenn eine Anfrage plötzlich zurückfällt. Du kannst den Hinweis sofort in deinem Anwendungscode einsetzen, um die Abfrage zu stabilisieren, während du die Ursache (fehlende Indizes, veraltete Statistiken oder Schemaänderungen) untersuchst.

Vermeiden Sie es, dauerhaft Hinweise zu hinterlassen. Wenn sich die Datenverteilung oder das Schema ändert, kann ein fest codierter Hinweis die Situation verschlimmern. Behandle sie als vorübergehend und überarbeite sie erneut, nachdem das zugrunde liegende Problem behoben ist:

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

Verwenden Sie OPTION (RECOMPILE), um schlechte zwischengespeicherte Pläne zu umgehen

SQL Server speichert Abfragepläne basierend auf dem ersten Satz von Parameterwerten, die er sieht. Wenn die Datenverteilung zwischen den Aufrufen stark variiert, kann der zwischengespeicherte Plan für einige Werte schlecht abschneiden. Dieses Problem, genannt Parameter-Sniffing, taucht oft als Abfrage auf, die "früher schnell war" und plötzlich Sekunden oder Minuten dauert.

OPTION (RECOMPILE)Zwingt SQL Server, für jede Ausführung einen neuen Plan zu erstellen, was eine effektive sofortige Lösung ist, die du ohne serverseitige Änderungen bereitstellen kannst. Der Nachteil sind geringe Kompilierungskosten pro Aufruf, doch bei Abfragen, die selten ausgeführt werden oder Ergebnismengen unterschiedlicher Größe zurückgeben, sind diese Kosten im Vergleich zur Ausführung eines schlechten Plans vernachlässigbar.

Sobald Sie das Problem stabilisiert haben, können Sie sich Zeit nehmen, eine dauerhafte Lösung anzuwenden, wie zum Beispiel die Abfrage neu zu schreiben, gefilterte Indizes hinzuzufügen oder Plan-Guides zu verwenden:

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

Leistungsüberwachung

Timen Sie Ihre Anfragen

Um langsame Operationen zu finden, umwickle Abfragen mit 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")

Für einen umfassenderen Überblick darüber, wo Ihre Anwendung Zeit verbringt, verwenden Sie das integrierte cProfile Modul von Python:

python -m cProfile -s cumtime my_app.py

Diese Ansicht zeigt die kumulierte Zeit pro Funktionsaufruf, was Ihnen hilft, festzustellen, ob Langsamkeit bei Abfrageausführung, Datenverarbeitung oder Netzwerklatenz liegt.

Verwenden Sie Abfragespeicher für serverseitige Analysen

Das clientseitige Timing zeigt an, wie lange eine Abfrage aus Sicht Ihrer Anwendung benötigt, kombiniert aber Netzwerklatenz, Serverausführungszeit und Clientverarbeitung. Abfragespeicher erfasst Ausführungspläne und Laufzeitstatistiken auf dem Server, sodass Sie genau sehen können, wie SQL Server jede Abfrage ausgeführt hat, wie oft sie ausgeführt wurde und wie sich die Leistung im Laufe der Zeit verändert hat.

Abfragespeicher ist besonders nützlich, um Parameter-Sniffing, Planregressionen und Abfragen zu identifizieren, die die meisten Serverressourcen verbrauchen. Sie können die Sichten sys.query_store_runtime_stats und sys.query_store_plan direkt abfragen oder die integrierten Abfragespeicher-Berichte in SQL Server Management Studio verwenden.

Verwenden Sie Performance Dashboard-Berichte

Die Performance Dashboard Reports im SQL Server Management Studio bieten einen Echtzeitüberblick über die Gesundheit von SQL Server, einschließlich aktueller Wartezeittypen, aktiven teuren Abfragen und CPU/IO-Trends. Nutze sie, um Engpässe schnell zu erkennen, ohne direkt Anfragen gegen DMVs zu stellen.

Prüfliste für die Leistung

Connection

• [ ] Aktiviere Verbindungspooling.
• [ ] Dimensioniere den Pool entsprechend deiner Arbeitsbelastung.
• [ ] Wiederverwenden von Verbindungen innerhalb der Operationen.
• [ ] Halte die Verbindungen in langjährigen Diensten offen.

Abfragen

• [ ] Wähle nur die Spalten aus, die du brauchst.
• [ ] Verwenden Sie für jede Abfrage die passende Abrufmethode.
• [ ] Implementiere serverseitige Paginierung.
• [ ] Nach dem Verbinden einmal SET NOCOUNT ON festlegen.
• [ ] Minimieren Sie Hin- und Rückläufe, indem Sie Abfragen bündeln.

Einfügungen

• [ ] Verwenden Sie execute() für einzeilige Einfügungen.
• [ ] Verwendung executemany() für kleine bis moderate Chargen (~10–1.000 Reihen).
• [ ] Nutze bulkcopy() , wenn Durchsatz wichtiger ist als Kontrolle pro Reihe.

Zwischenspeicherung

• [ ] Referenzdaten mit einer TTL zwischenspeichern, um zu vermeiden, dass veraltete Ergebnisse bereitgestellt werden.

Resources

• [ ] Verarbeiten Sie große Ergebnisse in Stücken oder mit Generatoren.
• [ ] Bereinigen Sie die Verbindungen umgehend.
• [ ] Überwachen Sie den Speicherverbrauch mit tracemalloc in lang laufenden Diensten.

Verwandte Inhalte

• Verbindungspooling
• Massenkopieroperationen
• Datenabruf