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가 소요될 수 있습니다. 시간당 수천 개의 메시지를 처리하는 큐 워커에게는 그 오버헤드가 빠르게 쌓입니다.
작업자가 평생 동안 연결을 열어두고, 연결이 끊기면 다시 연결하세요. 큐가 비어 있을 때 서버에 과부하가 가지 않도록 반복 사이에 슬립 모드(sleep)를 하세요:
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()
기본 모드인 연결 풀링이 활성화되면 풀이 유휴 연결을 처리합니다. 하지만 풀링을 비활성화하거나 단일 전용 연결을 사용하는 경우, 연결이 멈춰 있는 상태가 되지 않도록 오래된 연결을 조기에 감지할 수 있게 연결 문자열에서 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"}
)
적절한 삽입 방법을 선택하세요
드라이버는 서로 다른 스케일에 적합한 세 가지 데이터 삽입 방식을 제공합니다:
| 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() 더 뛰어나는지는 행 너비와 네트워크 지연 시간에 따라 다르지만, 보통 수백 행 초반 정도입니다. 아주 작은 배치의 경우, executemany() 별도의 내부 연결을 생성하고 자동으로 커밋하기 때문에 bulkcopy() 더 간단합니다.
execute() 및 executemany()와 달리 bulkcopy()는 INSERT 열 리스트가 아니라 위치를 기준으로 값을 열에 매핑합니다. 로드하는 목적지 열의 이름을 전달 column_mappings 하여 소스 튜플이 테이블의 앞 식별 열 대신 오른쪽 열과 일치하도록 합니다:
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)을 지원하지 않습니다. 연결당 활성 쿼리를 가질 수 있는 커서는 하나뿐입니다. 다음 쿼리를 실행하기 전에 첫 번째 결과 집합을 완전히 가져오거나, 두 번째 연결을 사용하세요:
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를 사용하여 서버 측에서 데이터를 페이지로 나누어 처리하고 한 번에 하나의 청크씩 처리하세요.
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()
메모리 사용량 모니터링
대규모 결과 집합, 장기 수명 캐시, 연결 객체 모두 메모리를 소모합니다. 애플리케이션이 서비스로 실행된다면, 닫히지 않은 커서나 경계가 없는 캐시로 인한 메모리 누수가 결국 운영체제나 컨테이너 런타임에 의해 프로세스가 종료될 수 있습니다.
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 ONSET 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는 각 실행마다 새로운 계획을 구축하도록 강요하며, 이는 서버 측 변경 없이도 배포할 수 있는 효과적이고 즉각적인 해결책입니다. 대가로 호출당 컴파일 비용이 적지만, 드물게 실행되거나 변수 크기의 결과 세트를 반환하는 쿼리의 경우, 나쁜 계획을 실행하는 것에 비하면 그 비용은 무시할 만합니다.
문제가 안정되면 쿼리 재작성, 필터링 인덱스 추가, 계획 가이드 사용 등 영구적인 해결책을 천천히 적용할 수 있습니다:
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에 직접 문의하지 않고도 병목 현상을 빠르게 파악할 수 있습니다.
성능 체크리스트
연결
- [ ] 연결 풀링 활성화.
- [ ] 업무량에 맞는 풀 크기를 정하세요.
- [ ] 운영 내에서 연결을 재사용하세요.
- [ ] 장시간 실행되는 서비스에서는 연결을 열린 상태로 유지하세요.
Queries
- [ ] 필요한 열만 선택하세요.
- [ ] 각 쿼리에 맞는 fetch 메서드를 사용하세요.
- [ ] 서버 측 페이지네이션을 구현하세요.
- [ ] 연결한 후
SET NOCOUNT ON을 한 번 설정하세요. - [ ] 쿼리를 배치하여 왕복을 최소화합니다.
삽입
- [ ] 단일 행 삽입에는
execute()을 사용합니다. - [ ] 소규모~중간 규모 배치(약 10~1,000행)에는
executemany()를 사용합니다. - [ ] 처리량이 행당 제어보다 더 중요할 때 사용
bulkcopy()하세요.
Caching
- [ ] TTL로 참조 데이터를 캐시하여 오래된 결과를 제공하지 않도록 하세요.
Resources
- [ ] 대용량 결과는 청크 단위로 또는 생성기를 사용해 처리하세요.
- [ ] 연결 부를 신속히 정리하세요.
- [ ] 장기 실행 서비스에서
tracemalloc메모리 사용량을 모니터링하세요.