이 글을 사용하여 쿼리 실행, 데이터 유형, 성능, 트랜잭션, 그리고 드라이버의 mssql-python 대량 복사 문제를 진단하세요.
쿼리 실행 문제
테이블 또는 객체를 찾지 못함
증상:
ProgrammingError: [42S02] (208) Invalid object name 'TableName'.
가능한 원인 및 해결 방법:
잘못된 데이터베이스 맥락
cursor.execute("SELECT DB_NAME()") print(cursor.fetchone()[0])스키마는 명시되지 않음
cursor.execute("SELECT * FROM dbo.TableName")테이블은 존재하지 않습니다
cursor.execute(""" SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'TableName' """)
구문 오류
증상:
ProgrammingError: [42000] (102) Incorrect syntax near '...'.
솔루션:
SQL Server Management Studio(SSMS)에서 SQL 문구를 테스트하여 문법을 검증하세요.
문자열 보간 대신 매개변수화된 쿼리를 사용하세요:
# Don't use string interpolation for query parameters. cursor.execute(f"SELECT * FROM Production.Product WHERE Name = '{name}'") # Use parameters. cursor.execute( "SELECT * FROM Production.Product WHERE Name = %(name)s", {"name": name}, )
매개변수 오류
증상:
ProgrammingError: [07001] Wrong number of parameters
솔루션:
자리 표시자와 매개변수를 세어보세요. 카운트가 일치해야 합니다.
올바른 매개변수 스타일을 선택하세요:
# Qmark style: positional parameters cursor.execute( "SELECT * FROM Production.Product " "WHERE ProductID = ? AND Name LIKE ?", (1, "Adjustable%"), ) print(cursor.fetchone()) # Pyformat style: named parameters cursor.execute( "SELECT * FROM Production.Product " "WHERE ProductID = %(id)s AND Name LIKE %(name)s", {"id": 1, "name": "Adjustable%"}, ) print(cursor.fetchone())
데이터 타입 문제
날짜 변환 오류
증상:
DataError: [22007] Invalid datetime format
Solution:
문자열 대신 Python datetime 객체를 사용하세요.
from datetime import datetime
cursor.execute("CREATE TABLE #Events (EventDate DATETIME)")
# This value raises an error because the date is invalid.
try:
cursor.execute(
"INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
{"event_date": "2024-13-45"},
)
except Exception as e:
print(f"Expected error: {e}")
# Use a Python datetime object.
cursor.execute(
"INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
{"event_date": datetime(2024, 3, 15)},
)
cursor.execute("SELECT EventDate FROM #Events")
print(cursor.fetchone())
소수점 정밀도 문제
증상:
숫자는 잘렸거나 잘못 반올림된 것처럼 보입니다.
Solution:
정확한 숫자 값 사용 decimal.Decimal :
from decimal import Decimal
cursor.execute("CREATE TABLE #PriceDemo (ListPrice DECIMAL(10, 2))")
cursor.execute(
"INSERT INTO #PriceDemo (ListPrice) VALUES (%(list_price)s)",
{"list_price": Decimal("19.99")},
)
유니코드 인코딩 문제
증상:
특수 문자가 잡음되거나 오류를 일으킵니다.
솔루션:
데이터베이스의 유니코드 데이터에는 nvarchar 컬럼을 사용하세요.
문자열을 직접 전달하세요. 드라이버는 인코딩을 담당합니다:
cursor.execute("CREATE TABLE #UnicodeDemo (Name NVARCHAR(50))") cursor.execute( "INSERT INTO #UnicodeDemo (Name) VALUES (%(name)s)", {"name": "日本語"}, ) cursor.execute("SELECT Name FROM #UnicodeDemo") print(cursor.fetchone())
성능 문제
느린 쿼리 실행
가능한 원인 및 해결 방법:
누락된 인덱스: SSMS에서 쿼리 실행 계획을 확인하세요.
대규모 결과 집합:
fetchall()대신fetchmany()사용:cursor.arraysize = 1000 while True: rows = cursor.fetchmany() if not rows: break process_rows(rows)연결 풀링 비활성화: 풀링 활성화:
import mssql_python mssql_python.pooling(max_size=20, idle_timeout=300)
큰 결과에서의 메모리 문제
증상:
Python 프로세스는 메모리가 부족합니다.
솔루션:
모든 행을 메모리에 불러오는 대신 스트림 결과를 선택하세요.
cursor.execute("SELECT * FROM LargeTable") for row in cursor: process_row(row)서버 측 페이지네이션을 사용하세요.
page_size = 1000 offset = 0 while True: cursor.execute( "SELECT * FROM LargeTable ORDER BY ID " "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY", (offset, page_size), ) rows = cursor.fetchall() if not rows: break process_rows(rows) offset += page_size
거래 문제
자동 커밋을 이용한 임시 테이블 스코핑
트랜잭션 내에서 생성한 세션 임시 테이블(#tablename)은 트랜잭션이 롤백될 때 사라집니다. 이 동작은 자동 커밋이 꺼져 있을 때 혼란을 일으키는 경우가 많으며, 자동 커밋이 기본값입니다:
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")
# An explicit rollback or an error removes #TempData.
conn.rollback()
# This statement fails with "Invalid object name '#TempData'".
cursor.execute("SELECT * FROM #TempData")
임시 테이블을 생성한 직후 바로 커밋하거나 자동 커밋 모드를 사용하세요:
cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
conn.commit()
cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")
conn.commit()
자동 커밋 모드가 필요한 DDL 문장들, 예: CREATE DATABASE, 는 열린 트랜잭션 내에서 실패합니다. 실행 전에 자동 커밋을 설정하세요:
conn.autocommit = True
cursor.execute("CREATE DATABASE TestDB")
conn.autocommit = False
트랜잭션 미완료
증상:
연결을 종료한 후에는 데이터 변경이 지속되지 않습니다.
Solution:
기본값인 autocommit=False를 사용하는 경우 commit()을 호출합니다:
cursor.execute("CREATE TABLE #Products (Name NVARCHAR(100))")
cursor.execute(
"INSERT INTO #Products (Name) VALUES (%(name)s)",
{"name": "Widget"},
)
conn.commit()
또는 자동 커밋 모드를 사용하세요:
conn = mssql_python.connect(connection_string, autocommit=True)
교착 상태 오류
증상:
OperationalError: [40001] (1205) Transaction ... was deadlocked on lock resources with another process
Solution:
재시도 로직 은 즉각적인 실패를 처리하지만, 반복되는 교착 상태는 설계 문제를 나타냅니다. 교착 상태 그래프를 캡처하고, 명제와 잠금 유형을 분석하세요. 일반적인 수정 방법으로는 다음과 같은 변경사항이 있습니다:
- 연산을 재정렬하여 경쟁 트랜잭션들이 같은 순서로 락을 획득하도록 합니다.
- 거래 범위를 줄이세요.
- 잠금 기간을 줄이기 위해 적절한 인덱스를 추가하세요.
교착 상태 분석의 전체 공략은 Deadlocks 가이드를 참조하세요. Azure SQL Database를 사용하는 경우 교착 상태를 분석하고 방지를 참고하세요.
벌크 로드 문제
대량 복사 중 제약 조건 위반
증상:
RuntimeError: CHECK constraint ... Conflict occurred in database ...
RuntimeError: Cannot insert duplicate key ... violation of PRIMARY KEY constraint
원인:
배치 내 데이터는 기본 키, 고유 키, 검사 키 또는 외래 키 제약 조건과 같은 테이블 제약 조건을 위반합니다.
Solution:
데이터를 로드하기 전에 반드시 검증하세요. 대규모 데이터셋의 경우, 데이터를 스테이징 테이블에 로드한 후 대상 테이블에 병합하세요:
cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(100))")
cursor.bulkcopy("##Staging", rows)
# Check for duplicate rows before the merge.
cursor.execute("""
SELECT s.ID
FROM ##Staging AS s
INNER JOIN dbo.Target AS t
ON s.ID = t.ID
""")
dupes = cursor.fetchall()
if dupes:
print(f"Skipping {len(dupes)} duplicate rows")
# Insert rows that don't exist in the target.
cursor.execute("""
INSERT INTO dbo.Target (ID, Name)
SELECT s.ID, s.Name
FROM ##Staging AS s
WHERE NOT EXISTS (
SELECT 1
FROM dbo.Target AS t
WHERE t.ID = s.ID
)
""")
conn.commit()
스테이징 테이블이 포함된 업서트 패턴에 대해서는 데이터 로딩 및 이동 패턴을 참조하세요.
열 매핑 오류
증상:
RuntimeError: Bulk copy failure - column count mismatch
원인:
데이터 내 열 수가 목표 테이블의 열 수와 일치하지 않거나, 열 순서가 잘못되어 있습니다.
Solution:
데이터가 순서와 횟수에서 테이블 스키마와 일치하는지 확인하세요:
from decimal import Decimal
cursor.execute("""
SELECT COLUMN_NAME, DATA_TYPE
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_NAME = 'MyTable'
ORDER BY ORDINAL_POSITION
""")
for col in cursor.fetchall():
print(col)
rows = [
(1, "Widget", Decimal("19.99")),
(2, "Gadget", Decimal("29.99")),
]
cursor.bulkcopy("dbo.MyTable", rows)
대량 복사 중 형식 불일치
증상:
데이터는 로드되지만 값이 잘려 나오거나 반올림되거나 잘못되어 있습니다.
원인:
Python 값이 대상 컬럼 타입에 깔끔하게 매핑되지 않습니다. 일반적인 예로는 정밀도가 손실될 수 있는 float 열에 로드된 값과 고정 길이 열에 로드된 너무 긴 문자열이 있습니다.
Solution:
스키마에 맞는 Python 타입을 사용하세요:
from decimal import Decimal
rows = [
# Use Decimal for decimal and numeric columns.
(1, "Widget", Decimal("19.99")),
# Avoid float values because they can lose precision.
# (1, "Widget", 19.99),
]
cursor.bulkcopy("dbo.Products", rows)
NumPy 타입 바인딩 실패
증상:
NumPy 정수나 float 타입을 사용할 때 매개변수가 조용히 실패하거나 데이터 타입 오류를 발생시킵니다.
원인:
NumPy 2.x에서는 numpy.int64 및 numpy.int32와 같은 NumPy 타입이 isinstance(x, int)를 통과하지 않습니다. 드라이버의 타입 추론이 이를 인식하지 못해 예상치 못한 행동을 유발합니다.
Solution:
NumPy 값을 바인딩하기 전에 네이티브 Python 타입으로 변환하세요:
import numpy as np
cursor.execute(
"SELECT * FROM Production.Product WHERE ProductID = %(product_id)s",
{"product_id": int(np.int64(42))},
)
for _, row in df.iterrows():
cursor.execute(
"INSERT INTO #Orders (ProductID, Qty) "
"VALUES (%(product_id)s, %(qty)s)",
{
"product_id": int(row["ProductID"]),
"qty": int(row["Qty"]),
},
)
더 큰 데이터셋의 경우, Arrow 나 pandas 통합 경로를 사용하세요. 이 경로들은 내부적으로 타입 변환을 처리합니다.
임시 테이블을 사용한 대량 복사
증상:
cursor.bulkcopy("#TempTable", data)에서 RuntimeError: Invalid object name '#TempTable'이 발생합니다.
원인:
bulkcopy() 메타데이터 조회 제한 때문에 세션 임시 테이블(#tablename)을 해결할 수 없습니다. 글로벌 온도 테이블(##tablename)과 영구 테이블이 작동합니다.
Solution:
글로벌 온도 테이블이나 일반 스테이징 테이블을 사용하세요:
# A global temp table is visible to all sessions and is dropped
# when the last session disconnects.
cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("##Staging", rows)
# Alternatively, use a permanent staging table.
cursor.execute("CREATE TABLE dbo.Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("dbo.Staging", rows)
세션 임시 테이블을 선호하는 작은 데이터셋의 경우, 다음을 사용하세요 executemany():
cursor.execute("CREATE TABLE #Staging (ID INT, Name NVARCHAR(50))")
cursor.executemany(
"INSERT INTO #Staging (ID, Name) VALUES (?, ?)",
rows,
)