이 기사에서는 인구 데이터를 분석하기 위해 SQL Server 데이터베이스와 상호 작용하는 Flask 기반 웹 애플리케이션의 개발 과정을 안내합니다. 이 애플리케이션은 사용자가 인구 범위를 조회하고 주별 카운티를 가져오며 특정 인구 범위 내의 주를 검색할 수 있게 합니다. 또한 쿼리 결과를 캐싱하여 성능을 향상시키기 위해 Redis를 통합하는 방법에 대해 논의할 것입니다.
왜 Flask, SQL Server 및 Redis를 사용하는가?
Flask는 소형에서 중형 웹 애플리케이션을 구축하는 데 적합한 가벼우면서 유연한 Python 웹 프레임워크입니다. RESTful API를 생성하고 동적 HTML 템플릿을 렌더링하며 데이터베이스와 상호 작용하는 데 필요한 도구를 제공합니다. 반면에, SQL Server는 기업 응용 프로그램에서 널리 사용되는 견고한 관계형 데이터베이스 관리 시스템(RDBMS)입니다. Flask와 SQL Server를 결합하면 데이터 분석 및 시각화를 위한 강력한 애플리케이션을 구축할 수 있습니다.
성능을 더 향상시키기 위해 Redis, 메모리 내 데이터 저장소를 통합하여 자주 액세스되는 쿼리 결과를 캐시합니다. 이를 통해 데이터베이스에 대한 부하를 줄이고 반복된 쿼리에 대한 응답 시간을 단축할 수 있습니다.
애플리케이션 개요
저희 Flask 애플리케이션은 다음 작업을 수행합니다:
- 쿼리 인구 범위. 사용자는 특정 연도와 인구 범위를 지정하여 해당 범위 내에 속하는 주 수를 얻을 수 있습니다.
- 주별 카운티 가져오기. 사용자는 주 코드를 입력하여 카운티 목록을 검색할 수 있습니다.
- 인구 범위별 주 검색. 사용자는 인구 범위와 연도를 지정하여 해당 범위 내에 있는 주 목록을 얻을 수 있습니다.
- 참고. 테스트하려면 데이터베이스에 자체 스키마를 만들고 다음 API를 사용하여 필요한 샘플 데이터를 삽입하십시오. 여기에서 사용되는 HTML 페이지는 Flask 앱 코드에서 반환된 데이터를 가져와 결과를 표시하는 기본 테이블 디자인일 수 있습니다.
구현 세부 정보로 들어가 봅시다.
Flask 애플리케이션 설정
1. 전제 조건
시작하기 전에 다음 사항이 터미널 루트를 통해 설치되어 있는지 확인하십시오 (MacOS와 호환되는 명령):
- Python 3.x
- Flask (
pip install flask) - SQLAlchemy (
pip install sqlalchemy) - PyODBC (
pip install pyodbc) - Redis (
pip install redis)
2. 데이터베이스 연결
우리는 SQL Server 데이터베이스에 연결하기 위해 SQLAlchemy를 사용합니다. 연결 구성 방법은 다음과 같습니다:
from sqlalchemy import create_engine
import urllib
# SQL Server connection string
params = urllib.parse.quote_plus(
"Driver={ODBC Driver 17 for SQL Server};"
"Server=tcp:username.database.windows.net,1433;"
"Database=population;"
"Uid=user@username;"
"Pwd={azure@123};"
"Encrypt=yes;"
"TrustServerCertificate=no;"
"Connection Timeout=30;"
)
engine = create_engine("mssql+pyodbc:///?odbc_connect=%s" % params)
이 연결 문자열은 SQL Server용 ODBC 드라이버를 사용하며 암호화 및 시간 초과에 대한 매개 변수를 포함합니다.
3. 레디스 구성
레디스는 쿼리 결과를 캐시하는 데 사용됩니다. 레디스 연결을 설정하는 방법은 다음과 같습니다:
import redis
# Redis connection
redis_client = redis.StrictRedis(
host='username.redis.cache.windows.net',
port=6380,
db=0,
password='encryptedpasswordstring',
ssl=True
)
4. 애플리케이션 라우트 구현
홈 페이지 라우트
홈 페이지 라우트는 애플리케이션의 메인 페이지를 렌더링합니다:
.route('/')
def index():
return render_template('index.html')
레디스 캐싱을 이용한 인구 범위 쿼리
이 라우트는 인구 범위에 대한 쿼리를 처리합니다. 먼저 결과가 레디스에 캐시되어 있는지 확인합니다. 그렇지 않으면 데이터베이스에 쿼리하고 결과를 캐시합니다:
.route('/population-range', methods=['GET', 'POST'])
def population_range():
if request.method == 'POST': # input params defined for this api
year = request.form['yr1']
range1_start = request.form['r1']
range1_end = request.form['r2']
range2_start = request.form['r3']
range2_end = request.form['r4']
range3_start = request.form['r5']
range3_end = request.form['r6']
# Map year to column name
year_map = {
'2010': 'ten',
'2011': 'eleven',
'2012': 'twelve',
'2013': 'thirteen',
'2014': 'fourteen',
'2015': 'fifteen',
'2016': 'sixteen',
'2017': 'seventeen',
'2018': 'eighteen'
}
year_column = year_map.get(year, 'ten') # Default to 'ten' if year not found
# Build cache key
cache_key = f"population_range_{year_column}_{range1_start}_{range1_end}_{range2_start}_{range2_end}_{range3_start}_{range3_end}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT
SUM(CASE WHEN {year_column} BETWEEN '{range1_start}' AND '{range1_end}' THEN 1 ELSE 0 END) AS range1_count,
SUM(CASE WHEN {year_column} BETWEEN '{range2_start}' AND '{range2_end}' THEN 1 ELSE 0 END) AS range2_count,
SUM(CASE WHEN {year_column} BETWEEN '{range3_start}' AND '{range3_end}' THEN 1 ELSE 0 END) AS range3_count
FROM popul
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('display.html', data1=result, t1=time_taken, cache_status=cache_status)
return render_template('index.html')
레디스 캐싱을 이용한 주별 카운티 조회
이 라우트는 주 코드에 대한 카운티를 검색합니다. 또한 결과를 캐시하기 위해 레디스를 사용합니다:
.route('/counties-by-state', methods=['GET', 'POST'])
def counties_by_state():
if request.method == 'POST':
state_code = request.form['state_code']
# Build cache key
cache_key = f"counties_by_state_{state_code}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT county
FROM dbo.county
WHERE state = (SELECT state FROM codes WHERE code = '{state_code}')
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('counties.html', data=result, time_taken=time_taken, cache_status=cache_status)
return render_template('index.html')
레디스 캐싱을 이용한 인구 범위별 주 조회
이 라우트는 지정된 인구 범위 내의 주를 검색하고 결과를 캐시합니다:
.route('/states-by-population', methods=['GET', 'POST'])
def states_by_population():
if request.method == 'POST':
year = request.form['year']
population_start = request.form['population_start']
population_end = request.form['population_end']
# Map year to column name
year_map = {
'2010': 'ten',
'2011': 'eleven',
'2012': 'twelve',
'2013': 'thirteen',
'2014': 'fourteen',
'2015': 'fifteen',
'2016': 'sixteen',
'2017': 'seventeen',
'2018': 'eighteen'
}
year_column = year_map.get(year, 'ten') # Default to 'ten' if year not found
# Build cache key
cache_key = f"states_by_population_{year_column}_{population_start}_{population_end}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT state
FROM popul
WHERE {year_column} BETWEEN '{population_start}' AND '{population_end}'
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('states.html', data=result, time_taken=time_taken, cache_status=cache_status)
return render_template('index.html')
성능 비교: SQL Server vs. 레디스
| Query Type | Redis Fetch Time | SQL Execution Time |
|---|---|---|
| 인구 범위 쿼리 (캐시됨) | 0.002 초 | 0.000 초 |
| 인구 범위 쿼리 (신선함) | 0.002 초 | 1.342 초 |
주요 요점: 레디스는 실행 시간을 약 1.3초에서 약 0.002초로 줄여 쿼리를 650배 더 빠르게 만듭니다!
Redis가 성능을 향상시키는 방법
Redis는 응용 프로그램과 데이터베이스 사이의 캐싱 레이어로 작동하는 인메모리 데이터 저장소입니다. 우리 응용 프로그램에서 Redis가 작동하는 방법은 다음과 같습니다:
- 캐시 키. 각 쿼리에 대해 고유한 키가 해당 매개변수를 기반으로 생성됩니다.
- 캐시 확인. 데이터베이스 쿼리를 실행하기 전에 응용 프로그램이 결과가 이미 Redis에 캐시되어 있는지 확인합니다.
- 캐시 히트. 결과가 Redis에 있는 경우 데이터베이스 쿼리를 회피하고 즉시 반환됩니다.
- 캐시 미스. 결과가 없는 경우 쿼리가 실행되고 결과가 미래 사용을 위해 Redis에 캐시됩니다.
- 캐시 만료. 캐시된 결과는 지정된 시간(예: 1시간) 후에 만료되도록 설정되어 데이터 신선도를 보장합니다.
Redis는 자주 액세스되는 쿼리 결과를 캐시함으로써 데이터베이스의 부하를 크게 줄이고 반복된 쿼리에 대한 응답 시간을 개선합니다.
결론
이 기사에서는 인구 데이터를 분석하기 위해 SQL Server 데이터베이스와 상호 작용하는 Flask 응용 프로그램을 구축했습니다. 우리는 쿼리 결과를 캐시하기 위해 Redis를 통합하여 성능을 향상시키고 데이터베이스 부하를 줄였습니다. 최상의 방법을 따르면이 응용 프로그램을 더 복잡한 쿼리를 처리하고 프로덕션 환경에서 확장할 수 있습니다.
링크: 이 전체 응용 프로그램의 소스 코드는 GitHub에서 찾을 수 있습니다.
Source:
https://dzone.com/articles/build-data-analytics-platform-flask-sql-redis