Ajuste de desempenho para aplicações mssql-python

O driver mssql-python oferece vários recursos e padrões para otimizar o desempenho das aplicações SQL Server, incluindo pooling de conexões, otimização de consultas e operações em massa.

Gerenciamento de conexões

Uso de agrupamento de conexões

O pool de conexões é nativo. Quando você chama conn.close(), a conexão retorna ao pool para ser reutilizada em vez de ser destruída, então as chamadas subsequentes a connect() evitam o handshake custoso:

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

Configure o tamanho do pool para a carga de trabalho

Ajuste o tamanho do grupo com base nos requisitos de concorrência. Se sua aplicação lida com muitos usuários simultâneos, aumente o pool. Para cargas de trabalho menores, um pool menor economiza recursos do servidor:

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
)

Reutilizar conexões dentro das operações

Abrir uma nova conexão para cada consulta adiciona sobrecarga, mesmo com pool de conexões. Em vez disso, mantenha uma única conexão durante a duração de uma operação lógica:

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

Mantenha conexões abertas em serviços de longa duração

Servidores web, trabalhadores de fila e trabalhos programados que rodam continuamente devem manter as conexões abertas em vez de se conectar e desconectar em todas as operações. Abrir uma conexão envolve um handshake TCP, negociação TLS e autenticação, que pode levar de 50 a 200 ms dependendo da distância da rede e do método de autenticação. Para um trabalhador da fila processando milhares de mensagens por hora, essa sobrecarga se acumula rapidamente.

Mantenha a conexão aberta durante toda a vida útil do trabalhador e reconecte quando a conexão falhar. Aguarde entre uma iteração e outra para evitar sobrecarregar o servidor quando a fila estiver vazia:

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

Com o pool de conexões ativado (o padrão), o pool cuida das conexões ociosas para você. Mas, se você desativar o pooling ou usar uma única conexão dedicada, defina Connection Timeout e Command Timeout na sua cadeia de conexão para detectar conexões obsoletas antecipadamente, em vez de travar.

Otimização de consultas

Busque apenas os dados necessários

Selecionar apenas as colunas que sua aplicação usa reduz a transferência de rede, o consumo de memória e o tempo de execução da consulta.

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

Use métodos de busca apropriados

O driver fornece vários métodos de obtenção. Use o que corresponde ao tamanho do seu resultado:

• fetchval() retorna um único valor escalar com sobrecarga mínima.
• fetchall() carrega todo o conjunto de resultados na memória, o que funciona bem para tabelas pequenas.
• fetchmany(n) recupera linhas em lotes, mantendo o uso de memória constante para grandes conjuntos de resultados.

O tamanho certo do lote fetchmany() depende da largura das linhas. Para linhas estreitas (algumas colunas pequenas, aproximadamente 1 KB cada), 1.000 linhas mantêm cada lote em torno de 1 MB de memória. Para fileiras largas com strings grandes ou colunas binárias, use um lote menor. Comece com 1.000 e ajuste com base nos seus dados.

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)

Use paginação do lado do servidor

Em vez de buscar todas as linhas e fatiar em Python, use OFFSET/FETCH NEXT para obter apenas a página necessária.

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

Uso SET NOCOUNT ON

Por padrão, o SQL Server envia uma mensagem "linhas afetadas" após cada instrução DML. SET NOCOUNT ON suprime essas mensagens e reduz o tráfego de rede. É uma configuração de nível de sessão, então defina uma vez após conectar, em vez de incorporá-la em todas as consultas.

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

Escolha o método certo de inserção

O driver oferece três maneiras de inserir dados, cada uma adequada para uma escala diferente:

Método	Contagem de linhas	Por que
execute()	1 linha por chamada	Use para operações de uma única linha, como envio de formulários ou manipuladores de API, quando você precisar imediatamente do ID inserido.
executemany()	~10-1.000 linhas	Usa vinculação de parâmetros por coluna para obter melhor taxa de transferência do que usar um loop. Envia cada linha como uma instrução parametrizada.
bulkcopy()	Centenas de linhas ou mais	Utiliza o protocolo TDS bulk insert, que é significativamente mais eficiente do que inserções linha por linha. Ideal para cargas de dados, migrações e processamento em lote.

Para mais detalhes e exemplos, veja Padrões de carregamento e movimento de dados.

Inserções individuais com execute()

Use para inserções pontuais quando você precisar do resultado imediatamente. Production.Product possui várias colunas NOT NULL sem valores padrão, então a instrução INSERT lista todas elas:

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

Inserções em lote com executemany()

executemany() vincula parâmetros por coluna e os envia de forma eficiente. Use-o para lotes de tamanho moderado em vez de chamar execute() em um loop. Note que executemany() requer marcadores posicionais ? com uma lista de tuplas, enquanto execute() suporta ambos ? e parâmetros nomeados %(name)s com dicts. Veja consultas parametrizadas para detalhes sobre cada estilo.

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

Cópia a granel para cargas grandes

Quando a taxa de transferência importa mais do que o controle por linha, mude para bulkcopy(). Ele transmite linhas pelo protocolo TDS bulk insert e evita o sobrecusto por linha das instruções parametrizadas. O ponto exato em que bulkcopy() supera executemany() depende da largura das linhas e da latência da rede, mas normalmente fica na faixa de pouco mais de cem linhas. Para lotes muito pequenos, executemany() é mais simples porque bulkcopy() cria uma conexão interna separada e confirma automaticamente.

Ao contrário de execute() e executemany(), bulkcopy() mapeia valores para as colunas por posição, não por uma lista de colunas INSERT. Passe column_mappings para especificar as colunas de destino para as quais você está carregando dados, para que as tuplas de origem se alinhem com as colunas corretas em vez da coluna de identidade no início da tabela:

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

Para cargas muito grandes, use um gerador para evitar carregar todo o conjunto de dados na memória e configure batch_size para fazer commit periodicamente:

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

Estratégias de cache

Para dados de referência que raramente mudam (categorias, tabelas de consulta, configuração), armazene os resultados em cache na sua aplicação em vez de consultar em todas as solicitações.

O functools.lru_cache do Python fornece memoização simples, mas mantém em cache indefinidamente até que o processo seja reiniciado. Se os dados subjacentes puderem mudar, use cachetools.TTLCache para atualizar automaticamente após um limite de tempo:

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

Otimização de rede

Minimizar viagens de ida e volta

Cada consulta é uma viagem de ida e volta à rede até o servidor. Combine consultas relacionadas em um único lote e use nextset() para avançar pelos conjuntos de resultados:

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

Use processamento do lado do servidor para lógica complexa

Faça agregação e filtragem para o SQL Server em vez de buscar linhas brutas e processá-las em Python. O servidor retorna uma única linha de resumo em vez de potencialmente milhares de linhas de detalhes:

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

Evite operações de cursor entrelaçado

O driver mssql-python não suporta Múltiplos Conjuntos de Resultados Ativos (MARS). Apenas um cursor pode ter uma consulta ativa por conexão. Busque o primeiro conjunto de resultados completamente antes de executar a próxima consulta, ou use uma segunda conexão:

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

Gerenciamento de memória

Processar grandes resultados em blocos

Carregar uma tabela de vários milhões de linhas em uma lista consome memória proporcional ao conjunto completo de resultados. Use OFFSET e FETCH NEXT para paginar os dados no lado do servidor e processar um bloco por vez.

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
)

Use geradores para streaming

Um gerador Python que envolve fetchmany() mantém o uso de memória constante, independentemente do tamanho da tabela. O chamador itera linha por linha sem carregar o conjunto completo de resultados. Para uma fonte extra-grande, combine tabelas com UNION ALL e transmita o resultado combinado da mesma forma.

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

Limpe os recursos rapidamente

Conexões não fechadas ocupam os recursos do servidor e podem esgotar o pool de conexões. Use um gerenciador de contexto para garantir a limpeza mesmo quando ocorrem exceções.

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

Monitorar o uso de memória

Grandes conjuntos de resultados, caches de longa duração e objetos de conexão consomem memória. Se sua aplicação é executada como um serviço, vazamentos de memória causados por cursores não fechados ou caches sem limites podem, com o tempo, fazer com que o processo seja encerrado pelo sistema operacional ou pelo ambiente de execução do contêiner.

Use o módulo tracemalloc do Python para capturar instantâneos da memória e encontrar as maiores alocações.

import tracemalloc

tracemalloc.start()

# ... run your workload ...

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

Fontes comuns de crescimento inesperado de memória incluem:

• Chamar fetchall() em uma consulta que retorna milhões de linhas. Use fetchmany() ou um gerador.
• Armazenamento em cache dos resultados da consulta sem um maxsize ou TTL. Os caches crescem até que o processo seja reiniciado.
• Criando cursores em um loop sem fechá-los. Cada cursor aberto mantém seu conjunto de resultados na memória.

Otimização de índice e plano de consulta

Verifique o desempenho das consultas do lado do servidor

Use SET STATISTICS TIME ON e SET STATISTICS IO ON para ver quanto tempo as consultas levam no servidor e quantos dados elas leem. Leituras lógicas altas geralmente indicam um índice ausente. Execute estas instruções no SQL Server Management Studio ou na extensão MSSQL para Visual Studio Code, onde a saída aparece no painel de Mensagens:

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

Você deverá ver uma saída como:

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

Se você observar um alto número de leituras lógicas ou varreduras de tabelas, considere adicionar um índice.

Use dicas de consulta como uma solução tática

As diretivas de consulta sobrescrevem as escolhas do otimizador de consultas quanto ao índice e à estratégia de junção. Em produção, eles são valiosos como um patch rápido e de baixo risco quando uma consulta regride de repente. Você pode implantar a dica no código da sua aplicação imediatamente para estabilizar a consulta enquanto investiga a causa raiz (índices faltando, estatísticas obsoletas ou mudanças de esquema).

Evite deixar pistas de forma permanente. Quando a distribuição dos dados ou o esquema muda, uma dica codificada pode piorar as coisas. Trate-os como temporários e revisite depois que o problema subjacente for resolvido:

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

Use a dica de consulta OPTION (RECOMPILE) para contornar planos ruins armazenados em cache

O SQL Server armazena em cache os planos de consulta com base no primeiro conjunto de valores de parâmetros que ele vê. Se a distribuição de dados variar bastante entre as chamadas, o plano em cache pode ter um desempenho ruim para alguns valores. Esse problema, chamado parameter sniffing, geralmente se manifesta quando uma consulta que "costumava ser rápida" de repente passa a levar segundos ou minutos.

OPTION (RECOMPILE)força o SQL Server a construir um novo plano para cada execução, o que é uma correção imediata e eficaz que você pode implantar sem nenhuma alteração do lado do servidor. O preço a pagar é um pequeno custo de compilação por chamada, mas, para consultas que são executadas com pouca frequência ou retornam conjuntos de resultados de tamanho variável, esse custo é negligenciável em comparação com a execução de um plano inadequado.

Depois de estabilizar o problema, você pode se dedicar a aplicar uma solução permanente, como reescrever a consulta, adicionar índices filtrados ou usar guias de planos:

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

Monitoramento de desempenho

Cronometre suas consultas

Para encontrar operações lentas, envolva consultas com 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")

Para uma visão mais ampla de onde sua aplicação gasta tempo, use o módulo embutido cProfile do Python:

python -m cProfile -s cumtime my_app.py

Esta visualização mostra o tempo acumulado por chamada de função, o que ajuda a identificar se a lentidão está na execução da consulta, processamento de dados ou latência de rede.

Use a Repositório de Consultas para análise do lado do servidor

O tempo do lado do cliente indica quanto tempo uma consulta leva do ponto de vista da sua aplicação, mas combina latência de rede, tempo de execução do servidor e processamento do cliente. A Repositório de Consultas captura planos de execução e estatísticas de tempo de execução no servidor, para que você possa ver exatamente como o SQL Server executou cada consulta, com que frequência foi executada e como seu desempenho mudou ao longo do tempo.

Repositório de Consultas é especialmente útil para identificar problemas de inspeção de parâmetros, regressões de plano e consultas que consomem mais recursos do servidor. Você pode consultar diretamente as exibições sys.query_store_runtime_stats e sys.query_store_plan ou usar os relatórios internos do Repositório de Consultas no SQL Server Management Studio.

Use os Relatórios do Painel de Desempenho

Os Relatórios do Painel de Desempenho no SQL Server Management Studio fornecem uma visão geral em tempo real da saúde do SQL Server, incluindo tipos de espera atuais, consultas ativas e caras e tendências de CPU/IO. Use-os para identificar rapidamente gargalos sem fazer consultas diretamente contra o DMV.

Lista de verificação de desempenho

Connection

• [ ] Ativar pool de conexões.
• [ ] Dimensione o pool para sua carga de trabalho.
• [ ] Reutilizar conexões dentro das operações.
• [ ] Mantenha conexões abertas em serviços de longa duração.

Queries

• [ ] Selecione apenas as colunas que precisa.
• [ ] Use o método de busca apropriado para cada consulta.
• [ ] Implementar paginação do lado do servidor.
• [ ] Defina SET NOCOUNT ON uma vez após conectar.
• [ ] Minimize viagens de ida e volta agrupando as consultas.

Inserções

• [ ] Use execute() para inserções de linha única.
• [ ] Use executemany() para lotes pequenos a médios (~10-1.000 linhas).
• [ ] Use bulkcopy() quando o throughput importa mais do que o controle por linha.

Cache

• [ ] Cache os dados de referência com TTL para evitar fornecer resultados obsoletos.

Resources

• [ ] Processe grandes resultados em blocos ou com geradores.
• [ ] Limpe as conexões rapidamente.
• [ ] Monitore o uso de memória com tracemalloc em serviços de longa execução.

Conteúdo relacionado

• Agrupamento de conexões
• Operações de cópia em massa
• Recuperação de dados