Always On 가용성 그룹은 SQL Server 데이터베이스에 대한 고가용성을 제공합니다. 연결 전에 데이터베이스 관리자는 가용성 그룹 리스너를 생성 하고 서버에 읽기 전용 라우팅을 설정 해야 합니다. mssql-python 드라이버는 다음 연결 문자열 키워드를 통해 가용성 그룹 연결을 지원합니다:
-
ApplicationIntent- 읽기 전용 보조 복제본으로 연결을 라우팅 -
MultiSubnetFailover- 더 빠른 장애 조치 를 위해 서브넷 간 병렬 연결 시도 활성화
가용성 그룹 리스너에 연결하세요
기본 리스너 연결
특정 서버 대신 리스너 DNS 이름을 통해 가용성 그룹에 연결하세요:
import mssql_python
# Connect via availability group listener
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
cursor = conn.cursor()
cursor.execute("SELECT @@SERVERNAME AS ServerName")
print(f"Connected to: {cursor.fetchval()}")
포트 사양 포함
리스너가 비기본 포트를 사용할 때 포트 번호를 지정하세요:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com,1433;" # Listener with port
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
읽기 전용 라우팅
읽기 전용 의도 활성화
보조 복제본으로 라우팅하려면 ApplicationIntent=ReadOnly을(를) 사용하세요:
# Connect for read operations - routes to secondary
read_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"Encrypt=yes;"
)
# Connect for read-write operations - routes to primary
write_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadWrite;" # Default
"Encrypt=yes;"
)
라우팅 검증
어떤 복제본이 연결을 처리했는지, 그리고 그것이 주 복제인지 보조 복제본인지 확인하세요:
def check_replica_role(conn) -> str:
"""Check if connected to primary or secondary."""
cursor = conn.cursor()
cursor.execute("""
SELECT
@@SERVERNAME AS ServerName,
CASE
WHEN DATABASEPROPERTYEX(DB_NAME(), 'Updateability') = 'READ_WRITE'
THEN 'Primary'
ELSE 'Secondary'
END AS Role
""")
row = cursor.fetchone()
return f"{row.ServerName} ({row.Role})"
read_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadOnly;")
print(f"Read connection: {check_replica_role(read_conn)}")
write_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadWrite;")
print(f"Write connection: {check_replica_role(write_conn)}")
다중 서브넷 장애 조치
멀티서브넷 장애 조치 활성화
여러 서브넷에 걸친 가용성 그룹의 경우:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
이 설정은 다음과 같습니다.
- 모든 IP 주소에 병렬로 연결을 시도합니다.
- 다중 서브넷 구성에서 장애 전환 시간을 줄입니다.
- 모든 가용성 그룹 연결에서 가장 잘 작동합니다.
연결 아키텍처 패턴
별도의 읽기 및 쓰기 연결
읽기 트래픽은 보조 회로, 쓰기는 주 회로로 가도록 별도의 연결을 유지하세요:
class DatabaseConnections:
"""Manage separate connections for read and write operations."""
def __init__(self, listener: str, database: str):
self.base_conn_str = f"Server={listener};Database={database};Authentication=ActiveDirectoryDefault;Encrypt=yes;"
self._read_conn = None
self._write_conn = None
@property
def read_connection(self):
"""Get or create read-only connection (secondary replica)."""
if self._read_conn is None:
self._read_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadOnly;MultiSubnetFailover=yes;"
)
return self._read_conn
@property
def write_connection(self):
"""Get or create read-write connection (primary replica)."""
if self._write_conn is None:
self._write_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadWrite;MultiSubnetFailover=yes;"
)
return self._write_conn
def close(self):
if self._read_conn:
self._read_conn.close()
if self._write_conn:
self._write_conn.close()
# Usage
db = DatabaseConnections("ag-listener.contoso.com", "<database>")
# Queries go to secondary
cursor = db.read_connection.cursor()
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
products = cursor.fetchall()
# Writes go to primary
cursor = db.write_connection.cursor()
cursor.execute("INSERT INTO #Products (Name) VALUES (%(name)s)", {"name": "New Product"})
db.write_connection.commit()
db.close()
읽기 후 쓰기 패턴
주 서버에 쓰고 보조 서버에 데이터를 검증한 후 복제 지연을 고려하세요:
def create_order_and_verify(conn_manager, order_data: dict):
"""Create order on primary, verify on secondary with eventual consistency."""
# Write to primary
write_cursor = conn_manager.write_connection.cursor()
write_cursor.execute("""
INSERT INTO #Orders (CustomerID, Total) VALUES (%(cust)s, %(total)s);
SELECT SCOPE_IDENTITY();
""", order_data)
order_id = write_cursor.fetchval()
conn_manager.write_connection.commit()
# Wait for replication (in production, use more sophisticated approach)
import time
time.sleep(1)
# Verify on secondary
read_cursor = conn_manager.read_connection.cursor()
read_cursor.execute("SELECT * FROM #Orders WHERE OrderID = %(id)s", {"id": order_id})
if read_cursor.fetchone():
print(f"Order {order_id} replicated to secondary")
else:
print(f"Order {order_id} not yet replicated")
return order_id
장애 조치(failover) 처리
연결 복원력
페일오버로 연결이 중단되면 자동으로 쿼리를 재시도합니다:
import time
def execute_with_failover_retry(conn_str: str, query: str, params: dict,
max_retries: int = 3) -> list:
"""Execute query with automatic reconnection on failover."""
for attempt in range(max_retries + 1):
try:
conn = mssql_python.connect(conn_str)
cursor = conn.cursor()
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return results
except mssql_python.OperationalError as e:
error_str = str(e)
# Check for failover-related errors
# Azure SQL transient error codes
# See: /azure/azure-sql/database/troubleshoot-common-errors-issues
if any(code in error_str for code in ["40613", "40197", "40501"]):
if attempt < max_retries:
print(f"Failover detected, retrying ({attempt + 1}/{max_retries})...")
time.sleep(5 * (attempt + 1)) # Exponential backoff
continue
raise
# Usage
results = execute_with_failover_retry(
"Server=<ag-listener>.contoso.com;Database=<database>;Authentication=ActiveDirectoryDefault;MultiSubnetFailover=yes;Encrypt=yes;",
"SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(cat)s",
{"cat": 5}
)
재시도 루프는 일시적 오류 코드에만 반응합니다. 인증 오류, 권한 오류, 쿼리 문법 오류와 같은 비일시적 실패는 재시도로 해결되지 않아 즉시 발생한다.
주요 역할 변화 감지
현재 복제 역할을 확인하고 주 복제본이 이동했는지 다시 연결하세요:
def is_primary(conn) -> bool:
"""Check if current connection is to primary replica."""
cursor = conn.cursor()
cursor.execute("""
SELECT DATABASEPROPERTYEX(DB_NAME(), 'Updateability') AS Updateability
""")
return cursor.fetchval() == 'READ_WRITE'
def ensure_primary(conn_str: str) -> mssql_python.Connection:
"""Ensure connection is to primary, reconnect if needed."""
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
if not is_primary(conn):
# Might happen during failover
conn.close()
time.sleep(2)
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
return conn
작업 부하 보고
보고서를 보조 장치로 오프로드
보조 복제본에 대해 보고 쿼리를 실행하여 주 복제본의 부하를 줄이세요:
class ReportingService:
"""Service that runs reports on secondary replicas."""
def __init__(self, listener: str, database: str):
self.conn_str = (
f"Server={listener};"
f"Database={database};"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
def run_report(self, report_query: str, params: dict = None) -> list:
"""Execute report query on secondary replica."""
conn = mssql_python.connect(self.conn_str)
cursor = conn.cursor()
try:
cursor.execute(report_query, params or {})
return cursor.fetchall()
finally:
cursor.close()
conn.close()
def get_sales_summary(self, start_date, end_date) -> dict:
"""Run sales summary report."""
results = self.run_report("""
SELECT
YEAR(OrderDate) AS Year,
MONTH(OrderDate) AS Month,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS TotalSales
FROM Sales.SalesOrderHeader
WHERE OrderDate BETWEEN %(start)s AND %(end)s
GROUP BY YEAR(OrderDate), MONTH(OrderDate)
ORDER BY Year, Month
""", {"start": start_date, "end": end_date})
return [
{"year": r.Year, "month": r.Month,
"orders": r.OrderCount, "sales": r.TotalSales}
for r in results
]
# Usage
reports = ReportingService("ag-listener.contoso.com", "sales_db")
summary = reports.get_sales_summary(date(2024, 1, 1), date(2024, 12, 31))
모범 사례
항상 MultiSubnetFailover를 사용하세요
더 빠른 장애 조치를 위해 모든 가용성 그룹 연결 문자열에 MultiSubnetFailover=yes를 포함하세요:
# Recommended for all AG connections
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;" # Always include this
"Encrypt=yes;"
)
의도를 작업에 맞추기
각 연산을 데이터를 읽거나 쓰는지 여부에 따라 올바른 복제본으로 라우팅하세요:
def get_appropriate_connection(operation: str, connections: DatabaseConnections):
"""Get connection appropriate for the operation type."""
read_operations = {"SELECT", "REPORT", "EXPORT", "ANALYTICS"}
if operation.upper() in read_operations:
return connections.read_connection
else:
return connections.write_connection
보조 핸들 일시 사용 불가
일시적인 오류가 발생하면 보조 복제본을 다시 시도한 후, 보조 복제본이 접근 불가능하면 주 복제본으로 다시 전환하세요.
import time
def query_with_fallback(conn_manager, query: str, params: dict, max_retries: int = 2):
"""Query the secondary, retrying transient errors before falling back to the primary."""
transient_codes = ["40613", "40197", "40501"]
for attempt in range(max_retries + 1):
try:
cursor = conn_manager.read_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()
except mssql_python.OperationalError as e:
# Re-raise failures that retrying can't fix, such as authentication,
# permission, or query syntax errors.
if not any(code in str(e) for code in transient_codes):
raise
# Retry the secondary for transient errors before giving up.
if attempt < max_retries:
time.sleep(2 * (attempt + 1))
continue
# Secondary still unavailable after retries: fall back to the primary.
print("Secondary unavailable, using primary")
cursor = conn_manager.write_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()