** AI는 사용하지 못하지만 학생들은 사용할 수 있는 Programming Language를 만들어 오면 Grade Letter를 바꿔드리겠습니다**

이때 드는 의문...

_      내가 할 수 있을까?_

프로그래밍 언어는 절차형 언어 *와 *함수형 언어로 나눌 수 있을 것이다. 그러나 둘의 차이점을 설명하기 이전에 둘 다 AI는 다룰 수 있다는 것인데, 그것은

고정된 규칙이 있고, 그것을 AI가 이해하기 때문이다.

따라서 개념적으로 보면, 내가 만들어야 하는 언어는

• 1. 고정된 규칙이 없다.
• 2. 규칙은 있으나 AI가 이해할 수 없다.

가 선택지가 될 것이다.

그러나

'고정된 규칙이 없다' 와 '프로그래밍 언어'가 양립할까?

혹은

AI가 이해 못하는 규칙은 자연어로 이루어진 룰북이 없어야 하는가?

라는 고민에 도달하게 되었다.

솔직히 안될 거 같은데, 시간이 나면 클로드 잡도리로 클로드가 못 푸는 언어를 만들어 봐야겠다..

1편에서 250개의 쿼리를 1개로 줄이는 최적화를 진행했습니다.

-- 250개 좌표에 대해 LATERAL JOIN으로 한 번에 조회
WITH target_coords AS (
    SELECT * FROM (VALUES ('11010', 37.5, 127.0), ...)
    AS t(region_code, target_lat, target_lng)
)
SELECT DISTINCT ON (tc.region_code, hr.target_year)
    tc.region_code,
    hr.target_year,
    hr.score
FROM target_coords tc
CROSS JOIN LATERAL (
    SELECT target_year, score
    FROM hazard_results
    WHERE risk_type = %s
    AND target_year BETWEEN %s AND %s  -- 예: 2025 ~ 2100
    ORDER BY (거리 계산) ASC
    LIMIT 1
) hr
ORDER BY tc.region_code, hr.target_year

성능은 극적으로 개선되었지만... 새로운 문제가 발생했습니다.

문제 발견: 각 지역이 1개 연도 데이터만 반환

요구사항

• 각 행정구역마다 2025, 2030, 2035, ..., 2095, 2100년 (16개 연도) 데이터 필요
• 클라이언트는 연도별 시계열 그래프를 그려야 함

실제 결과

{
  "regionScores": {
    "11010": { "2025": 33.1 },  // ❌ 2025년만!
    "11020": { "2025": 30.9 },  // ❌ 2025년만!
    "11030": { "2025": 33.5 }   // ❌ 2025년만!
  }
}

기대한 결과

{
  "regionScores": {
    "11010": {
      "2025": 33.1,
      "2030": 35.2,
      "2035": 37.5,
      // ... 중략
      "2100": 55.8
    }
  }
}

원인 분석: LATERAL JOIN의 동작 방식

문제의 쿼리 구조

FROM target_coords tc              -- 250개 행
CROSS JOIN LATERAL (
    SELECT target_year, score
    FROM hazard_results
    WHERE risk_type = 'extreme_heat'
    AND target_year BETWEEN '2025' AND '2100'  -- 🔴 여기가 문제
    ORDER BY (거리 계산) ASC
    LIMIT 1                        -- 🔴 각 지역당 1개만!
) hr

LATERAL JOIN의 동작:

1. target_coords의 각 행(지역)마다 서브쿼리 실행
2. BETWEEN '2025' AND '2100' 범위 내에서
3. 거리 기준으로 정렬해서
4. LIMIT 1로 가장 가까운 데이터 1개만 선택

문제점:

• LATERAL은 250개 지역에 대해서만 반복 실행
• 각 지역마다 연도 구분 없이 가장 가까운 데이터 1개만 가져옴
• 그 1개가 우연히 2025년이었던 것!

비유로 이해하기

현재 쿼리:
"각 지역마다, 2025~2100년 중에서 가장 가까운 데이터 1개만 줘"
→ 결과: 지역당 1개 (대부분 2025년)

원하는 동작:
"각 지역마다, 각 연도별로 가장 가까운 데이터를 줘"
→ 결과: 지역당 16개 (연도별로)

해결 방법 1: 연도도 CROSS JOIN으로 추가

핵심 아이디어

• 지역만 반복하는 게 아니라
• 지역 × 연도 모든 조합에 대해 LATERAL JOIN 실행

수정된 쿼리

WITH target_coords AS (
    -- 250개 지역 좌표
    SELECT * FROM (VALUES
        ('11010', 37.5, 127.0),
        ('11020', 37.6, 127.1),
        -- ... 250개
    ) AS t(region_code, target_lat, target_lng)
),
target_years AS (
    -- 16개 연도 (NEW!)
    SELECT * FROM (VALUES
        ('2025'), ('2030'), ('2035'), ('2040'),
        ('2045'), ('2050'), ('2055'), ('2060'),
        ('2065'), ('2070'), ('2075'), ('2080'),
        ('2085'), ('2090'), ('2095'), ('2100')
    ) AS y(year)
)
SELECT DISTINCT ON (tc.region_code, ty.year)
    tc.region_code,
    ty.year as target_year,
    hr.score
FROM target_coords tc
CROSS JOIN target_years ty          -- 🟢 연도도 CROSS JOIN!
CROSS JOIN LATERAL (
    SELECT score
    FROM hazard_results
    WHERE risk_type = %s
    AND target_year = ty.year       -- 🟢 특정 연도만 조회
    ORDER BY (
        POW(latitude - tc.target_lat::numeric, 2) +
        POW(longitude - tc.target_lng::numeric, 2)
    ) ASC
    LIMIT 1                         -- 🟢 해당 연도의 최근접 데이터 1개
) hr
ORDER BY tc.region_code, ty.year

동작 방식

Before:
┌─────────┐
│ 지역 250개 │ → LATERAL → 각 지역당 1개 데이터
└─────────┘

After:
┌─────────┐   ┌────────┐
│ 지역 250개 │ × │ 연도 16개 │ → 250 × 16 = 4,000개 조합
└─────────┘   └────────┘
      ↓
각 (지역, 연도) 조합마다 LATERAL 실행
      ↓
4,000개 데이터 (지역당 16개 연도)

코드 변경 내용

Before (1개 연도만 반환)

query_region_batch = f"""
    WITH target_coords AS (
        SELECT * FROM (VALUES {coords_clause})
        AS t(region_code, target_lat, target_lng)
    )
    SELECT DISTINCT ON (tc.region_code, hr.target_year)
        tc.region_code,
        hr.target_year,
        hr.{score_col} as score
    FROM target_coords tc
    CROSS JOIN LATERAL (
        SELECT target_year, {score_col}
        FROM hazard_results
        WHERE risk_type = %s
        AND target_year BETWEEN %s AND %s
        ORDER BY (거리 계산) ASC
        LIMIT 1
    ) hr
    ORDER BY tc.region_code, hr.target_year
"""

After (모든 연도 반환)

# 고정된 연도 범위 생성
fixed_years = list(range(2025, 2101, 5))  # [2025, 2030, ..., 2100]

query_region_batch = f"""
    WITH target_coords AS (
        SELECT * FROM (VALUES {coords_clause})
        AS t(region_code, target_lat, target_lng)
    ),
    target_years AS (
        SELECT * FROM (VALUES {','.join("('" + str(y) + "')" for y in fixed_years)})
        AS y(year)
    )
    SELECT DISTINCT ON (tc.region_code, ty.year)
        tc.region_code,
        ty.year as target_year,
        hr.{score_col} as score
    FROM target_coords tc
    CROSS JOIN target_years ty
    CROSS JOIN LATERAL (
        SELECT {score_col}
        FROM hazard_results
        WHERE risk_type = %s
        AND target_year = ty.year
        ORDER BY (
            POW(latitude - tc.target_lat::numeric, 2) +
            POW(longitude - tc.target_lng::numeric, 2)
        ) ASC
        LIMIT 1
    ) hr
    ORDER BY tc.region_code, ty.year
"""

추가로 발견한 문제: 타입 불일치

에러 발생

psycopg2.errors.UndefinedFunction: operator does not exist: character varying = integer
LINE 9: AND target_year IN (2025,2030,2035,...)
        ^
HINT: No operator matches the given name and argument types.
You might need to add explicit type casts.

원인

• DB의 target_year 컬럼: VARCHAR (문자열)
• 쿼리의 값: 2025 (정수)
• PostgreSQL은 자동 타입 변환을 하지 않음

해결

# Before (정수)
AND target_year IN ({','.join(str(y) for y in fixed_years)})
# 결과: AND target_year IN (2025,2030,2035,...)

# After (문자열로 감싸기)
AND target_year IN ({','.join("'" + str(y) + "'" for y in fixed_years)})
# 결과: AND target_year IN ('2025','2030','2035',...)

성능 비교

LATERAL JOIN 사용 시 주의사항

1. 반복 단위를 명확히 하기

-- ❌ 잘못된 생각: "LATERAL이 알아서 연도별로 반복하겠지"
CROSS JOIN LATERAL (
    WHERE target_year BETWEEN '2025' AND '2100'
    LIMIT 1
)

-- ✅ 올바른 방법: "반복할 것을 명시적으로 CROSS JOIN"
CROSS JOIN target_years ty
CROSS JOIN LATERAL (
    WHERE target_year = ty.year
    LIMIT 1
)

2. LIMIT의 의미 이해하기

-- LIMIT 1의 의미:
-- "각 LATERAL 실행마다 1개만 반환"
-- ≠ "전체 결과 중 1개만 반환"

-- 지역 250개 × LATERAL LIMIT 1 = 250개 결과
-- 지역 250개 × 연도 16개 × LATERAL LIMIT 1 = 4,000개 결과

3. DISTINCT ON 활용

SELECT DISTINCT ON (tc.region_code, ty.year)
    -- (지역, 연도) 조합마다 첫 번째 행만 선택
    -- 이미 LATERAL에서 LIMIT 1을 했으므로 중복 방지용

최종 결과

API 응답 예시

{
  "regionScores": {
    "11010": {
      "2025": 33.1,
      "2030": 35.4,
      "2035": 38.2,
      "2040": 41.5,
      "2045": 44.8,
      "2050": 48.1,
      "2055": 51.6,
      "2060": 54.9,
      "2065": 58.3,
      "2070": 61.8,
      "2075": 65.2,
      "2080": 68.7,
      "2085": 72.1,
      "2090": 75.6,
      "2095": 79.0,
      "2100": 82.5
    },
    "11020": { /* 16개 연도 */ },
    // ... 248개 지역 더
  },
  "siteAALs": {
    "uuid1": { /* 16개 연도 */ },
    "uuid2": { /* 16개 연도 */ }
  }
}

학습 포인트

1. LATERAL JOIN은 만능이 아니다

• 왼쪽 테이블의 각 행마다 서브쿼리 실행
• 다차원 반복이 필요하면 명시적으로 CROSS JOIN

2. SQL은 명시적이어야 한다

• "DB가 알아서 해주겠지" ❌
• "내가 원하는 걸 정확히 표현" ✅

3. 쿼리 결과를 항상 검증

# 단순히 쿼리가 성공했다고 끝이 아니라
region_rows = db.execute_query(...)

# 결과 개수를 확인
expected_count = len(REGION_COORD_MAP) * len(fixed_years)
actual_count = len(region_rows)
assert actual_count == expected_count, \
    f"Expected {expected_count}, got {actual_count}"

결론

LATERAL JOIN을 사용할 때는:

1. 반복 단위를 명확히: 무엇을 기준으로 반복할 것인가?
2. CROSS JOIN으로 명시: 다차원 반복은 명시적으로 표현
3. LIMIT의 범위 이해: 각 LATERAL 실행마다의 제한
4. 결과 검증: 기대한 데이터 개수가 맞는지 확인

250개 쿼리를 1개로 줄이는 것도 중요하지만, 올바른 결과를 반환하는 것이 더 중요합니다.

참고 자료

• PostgreSQL CROSS JOIN
• LATERAL JOIN 상세 예제
• Common Table Expression (CTE) Best Practices

FastAPI 기후 시뮬레이션 API에서 30초 타임아웃 문제가 발생했습니다.

• 250개 행정구역 × 80년 × 4개 시나리오 × 9개 리스크 데이터 조회
• API 응답 시간: 30초 이상
• 결과: 타임아웃으로 실패

원인 분석

# 기존 코드 (문제)
for code, coord in REGION_COORD_MAP.items():  # 250번 반복
    target_lat = coord["lat"]
    target_lng = coord["lng"]

    query = """
        SELECT DISTINCT ON (target_year)
            target_year, score, latitude, longitude
        FROM hazard_results
        WHERE risk_type = %s
        AND target_year BETWEEN %s AND %s
        ORDER BY target_year, (
            POW(latitude - %s, 2) + POW(longitude - %s, 2)
        ) ASC
    """

    # 매번 DB에 쿼리 실행
    region_rows = db.execute_query(query, (risk_type, start_year, end_year, lat, lng))

문제점:

• Python 루프에서 250번 DB 쿼리 실행
• 250번의 네트워크 왕복 (Round-trip)
• 각 쿼리마다 연결, 파싱, 실행 오버헤드

해결 방법: LATERAL JOIN을 활용한 배치 쿼리

핵심 아이디어

1. 250개 좌표를 SQL VALUES로 임시 테이블 생성
2. LATERAL JOIN으로 각 좌표별 최근접 데이터 한 번에 조회
3. 1번의 쿼리로 모든 데이터 처리

개선된 코드

# 1. 250개 좌표를 VALUES 절로 변환
coords_values = []
for code, coord in REGION_COORD_MAP.items():
    coords_values.append(f"('{code}', {coord['lat']}, {coord['lng']})")

coords_clause = ', '.join(coords_values)
# 결과: "('11010', 37.5, 127.0), ('11020', 37.6, 127.1), ..."

# 2. 단일 쿼리로 모든 지역의 모든 연도 데이터 조회
query_region_batch = f"""
    WITH target_coords AS (
        SELECT * FROM (VALUES {coords_clause})
        AS t(region_code, target_lat, target_lng)
    )
    SELECT DISTINCT ON (tc.region_code, hr.target_year)
        tc.region_code,
        hr.target_year,
        hr.{score_col} as score
    FROM target_coords tc
    CROSS JOIN LATERAL (
        SELECT target_year, {score_col}
        FROM hazard_results
        WHERE risk_type = %s
        AND target_year BETWEEN %s AND %s
        ORDER BY (
            POW(latitude - tc.target_lat::numeric, 2) +
            POW(longitude - tc.target_lng::numeric, 2)
        ) ASC
        LIMIT 1
    ) hr
    ORDER BY tc.region_code, hr.target_year
"""

# 한 번만 실행
region_rows = db.execute_query(
    query_region_batch,
    (request.hazard_type, str(request.start_year), str(request.end_year))
)

핵심 기술 설명

1. CTE (Common Table Expression)

WITH target_coords AS (
    SELECT * FROM (VALUES
        ('11010', 37.5, 127.0),
        ('11020', 37.6, 127.1),
        -- ... 250개
    ) AS t(region_code, target_lat, target_lng)
)

• 250개 좌표를 메모리상의 임시 테이블로 생성
• 쿼리 내에서 여러 번 참조 가능

2. LATERAL JOIN

FROM target_coords tc
CROSS JOIN LATERAL (
    SELECT ...
    FROM hazard_results
    WHERE ...
    ORDER BY 거리계산
    LIMIT 1
) hr

일반 JOIN vs LATERAL JOIN:

LATERAL의 장점:

• 250개 좌표 각각에 대해 최근접 hazard 데이터를 찾음
• DB 엔진이 병렬 처리 최적화
• Python 루프보다 훨씬 효율적

3. DISTINCT ON

SELECT DISTINCT ON (tc.region_code, hr.target_year)
    tc.region_code,
    hr.target_year,
    hr.score
FROM ...
ORDER BY tc.region_code, hr.target_year

• (지역코드, 연도) 조합마다 첫 번째 행만 선택
• 거리 기준 정렬 후 가장 가까운 데이터만 추출

비유로 이해하기

기존 방식 (250번 왕복)

개발자: "서울 종로구 근처 데이터 찾아줘"
DB: "찾았어요"
개발자: "서울 중구 근처 데이터 찾아줘"
DB: "찾았어요"
... 248번 더 반복

개선된 방식 (1번 왕복)

개발자: "이 250개 좌표 근처 데이터를 한번에 찾아줘"
        [서울 종로구, 서울 중구, ... 250개]
DB: "250개 전부 찾아서 한 번에 줄게"

성능 개선 결과

추가 최적화: 인덱스 생성

-- 복합 인덱스로 검색 속도 향상
CREATE INDEX idx_hazard_results_risk_year_coords
ON hazard_results (risk_type, target_year, latitude, longitude);

-- 좌표 기반 검색 최적화
CREATE INDEX idx_hazard_results_coords_btree
ON hazard_results (latitude, longitude);

적용 가능한 상황

이 패턴은 다음과 같은 경우에 유용합니다:

1. 다중 좌표 최근접 검색

   • 여러 위치의 가장 가까운 매장/시설 찾기
   • 지역별 날씨/환경 데이터 조회

2. 배치 데이터 조회

   • N개 ID에 대한 관련 데이터 조회
   • 각 항목마다 조건부 서브쿼리 필요한 경우

3. Python 루프 → SQL 변환

   • 반복문에서 매번 DB 쿼리하는 경우
   • N+1 쿼리 문제 해결

주의사항

1. 좌표 개수 제한

# 너무 많은 좌표는 쿼리 크기 초과 가능
if len(REGION_COORD_MAP) > 1000:
    # 배치를 나눠서 처리
    batch_size = 500
    for i in range(0, len(coords), batch_size):
        batch = coords[i:i+batch_size]
        # 배치별 쿼리 실행

2. SQL Injection 방지

# BAD: f-string으로 사용자 입력 직접 삽입
coords_clause = f"('{user_input}', ...)"  # 위험!

# GOOD: 고정된 데이터만 VALUES에 사용
# 동적 파라미터는 %s 사용

3. 인덱스 활용

• ORDER BY POW(...) 계산은 인덱스 사용 불가
• 데이터가 많으면 좌표 범위로 필터링 후 거리 계산

  WHERE latitude BETWEEN %s - 0.1 AND %s + 0.1
  AND longitude BETWEEN %s - 0.1 AND %s + 0.1

전체 코드 비교

Before (250개 쿼리)

region_scores_map = {}

for code, coord in REGION_COORD_MAP.items():
    target_lat = coord["lat"]
    target_lng = coord["lng"]

    query_region = f"""
        SELECT DISTINCT ON (target_year)
            target_year, {score_col} as score
        FROM hazard_results
        WHERE risk_type = %s
        AND target_year BETWEEN %s AND %s
        ORDER BY target_year, (
            POW(latitude - %s, 2) + POW(longitude - %s, 2)
        ) ASC
    """

    region_rows = db.execute_query(
        query_region,
        (hazard_type, start_year, end_year, target_lat, target_lng)
    )

    if code not in region_scores_map:
        region_scores_map[code] = {}

    for row in region_rows:
        year = str(row['target_year'])
        score = float(row['score'] or 0.0)
        region_scores_map[code][year] = score

After (1개 쿼리)

region_scores_map = {}

if REGION_COORD_MAP:
    # 1. 좌표 값들을 VALUES 절로 변환
    coords_values = []
    for code, coord in REGION_COORD_MAP.items():
        coords_values.append(f"('{code}', {coord['lat']}, {coord['lng']})")

    coords_clause = ', '.join(coords_values)

    # 2. 단일 쿼리로 모든 지역의 모든 연도 데이터 조회
    query_region_batch = f"""
        WITH target_coords AS (
            SELECT * FROM (VALUES {coords_clause})
            AS t(region_code, target_lat, target_lng)
        )
        SELECT DISTINCT ON (tc.region_code, hr.target_year)
            tc.region_code,
            hr.target_year,
            hr.{score_col} as score
        FROM target_coords tc
        CROSS JOIN LATERAL (
            SELECT target_year, {score_col}
            FROM hazard_results
            WHERE risk_type = %s
            AND target_year BETWEEN %s AND %s
            ORDER BY (
                POW(latitude - tc.target_lat::numeric, 2) +
                POW(longitude - tc.target_lng::numeric, 2)
            ) ASC
            LIMIT 1
        ) hr
        ORDER BY tc.region_code, hr.target_year
    """

    region_rows = db.execute_query(
        query_region_batch,
        (hazard_type, start_year, end_year)
    )

    # 결과를 region_scores_map에 저장
    for row in region_rows:
        code = row['region_code']
        year = str(row['target_year'])
        score = float(row['score'] or 0.0)

        if code not in region_scores_map:
            region_scores_map[code] = {}
        region_scores_map[code][year] = score

결론

Python 루프에서 반복적으로 DB 쿼리를 실행하는 대신, SQL의 강력한 기능(CTE, LATERAL JOIN)을 활용하면:

• 네트워크 왕복 최소화
• DB 엔진의 최적화 활용
• 극적인 성능 향상

핵심 원칙: "데이터를 애플리케이션으로 가져와서 처리하지 말고, DB에서 처리해서 결과만 가져오자"

참고 자료

• PostgreSQL LATERAL JOIN 공식 문서
• Common Table Expressions (CTE)
• DISTINCT ON

배치 작업 실행 시 다음과 같은 경고가 발생하며 작업이 skip됨:

2025-12-18 09:13:54 - WARNING - Execution of job "P(H) Batch (Custom Trigger)" skipped:
maximum number of running instances reached (1)

특이사항:

• 실제로 배치 계산이 실행 중이지 않음
• ps aux 확인 결과 배치 프로세스가 없음
• 하지만 APScheduler는 여전히 인스턴스가 실행 중이라고 판단
• 새로운 배치 작업을 실행할 수 없음

원인 분석

APScheduler의 Instance 관리 메커니즘

APScheduler는 각 Job 함수의 실행 인스턴스 수를 내부적으로 추적합니다:

# Job 실행 전
job._instances += 1
if job._instances > job.max_instances:
    # "maximum number of running instances reached" 경고
    job._instances -= 1
    return

# Job 실행 완료 후 (finally 블록)
try:
    job.func(*args, **kwargs)
except:
    # 에러 로깅
finally:
    job._instances -= 1  # 반드시 실행되어야 함

핵심: job._instances 카운트는 Job 함수가 return될 때만 감소합니다.

실제 발생한 문제

1. ProcessPoolExecutor가 Hang 상태에 빠짐

배치 코드 구조:

# probability_timeseries_batch.py
def run_probability_batch(...):
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(_process_task_worker, task): task
            for task in tasks
        }

        for future in as_completed(futures):
            result = future.result()  # ❌ Timeout 없음!
            # 결과 처리...

문제 시나리오:

1. 배치 시작 (00:17:10.053)
   ↓
2. ProcessPoolExecutor 생성, 4개 worker 프로세스 시작
   ↓
3. Worker에서 DB 연결 시도
   ↓
4. DB 연결 실패: "connection already closed" (00:17:10.110)
   - Connection Pool Race Condition 발생
   - Worker 프로세스들이 무효한 연결을 받음
   ↓
5. Worker가 DB 재연결을 시도하며 무한 대기
   ↓
6. future.result()가 무한 대기 (Timeout 없음)
   ↓
7. ProcessPoolExecutor의 with 블록이 종료되지 않음
   ↓
8. run_probability_batch() 함수가 return되지 않음
   ↓
9. probability_batch_job() 함수도 return되지 않음
   ↓
10. APScheduler는 _instances 카운트를 감소시키지 못함
    ↓
11. 영구적으로 "maximum instances reached" 상태 유지

2. 증거 확인

배치 시작 로그는 있지만 종료 로그가 없음:

# main.py
def probability_batch_job():
    logger.info("P(H) BATCH JOB STARTED")  # ✓ 로그 있음

    try:
        run_probability_batch(...)
        logger.info("P(H) BATCH JOB COMPLETED SUCCESSFULLY")  # ❌ 로그 없음
    except Exception as e:
        logger.error(f"P(H) BATCH JOB FAILED: {e}")  # ❌ 로그 없음

결론: 함수가 try 블록 내에서 멈춰서 완료 로그도, 에러 로그도 출력되지 않음

3. 프로세스 상태 확인

# 컨테이너 내부에서 확인
ps aux | grep -E "(probability|hazard)"
# → 배치 프로세스 없음

ps -eLf | grep python | wc -l
# → 24개 Python 쓰레드 (main + API workers)

# 00:57-00:58에 생성된 worker 프로세스들 발견
# 이들은 site_assessment API의 ThreadPoolExecutor 워커들

해결 방법

1. Timeout 추가 (적용한 해결책)

개별 Task Timeout

각 태스크(격자점 하나의 계산)가 5분 안에 완료되지 않으면 실패 처리:

# probability_timeseries_batch.py (Line 271)
# hazard_timeseries_batch.py (Line 297)

for future in as_completed(futures):
    task = futures[future]
    try:
        result = future.result(timeout=300)  # ✓ 5분 timeout 추가

        if result['status'] == 'success':
            # 결과 처리...
        else:
            failed_count += 1

    except TimeoutError:
        # Timeout 발생 시 해당 태스크만 실패 처리
        failed_count += 1
        logger.error(f"Task timeout after 300s: {task}")
        # 다음 태스크 계속 진행

효과:

• Worker가 멈춰도 5분 후 해당 태스크만 실패 처리
• ProcessPoolExecutor가 무한 대기하지 않음
• run_probability_batch() 함수가 항상 return됨
• APScheduler 인스턴스 카운트가 정상적으로 감소
• 다음 배치 스케줄 가능

적용 전후 비교

Before:

✗ Worker 멈춤 → future.result() 무한 대기
✗ 함수 return 안 됨 → 인스턴스 release 안 됨
✗ 다음 배치 skip: "maximum instances reached"

After:

✓ Worker 멈춤 → 5분 후 TimeoutError
✓ 해당 태스크 실패 처리, 다음 태스크 계속
✓ 함수 항상 return → 인스턴스 정상 release
✓ 다음 배치 정상 실행 가능

추가 방어 방안 (프로덕션 환경)

2. 전체 Batch Timeout

전체 배치 실행 시간 제한:

# as_completed에 전체 timeout 추가
for future in as_completed(futures, timeout=7200):  # 2시간
    try:
        result = future.result(timeout=300)  # 개별 5분

3. Circuit Breaker 패턴

연속 실패 시 빠른 실패 처리:

class ConnectionCircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.last_failure_time = None
        self.timeout = timeout

    def execute(self, func):
        # Circuit이 열린 상태인지 확인
        if self.failures >= self.threshold:
            if (datetime.now() - self.last_failure_time).seconds < self.timeout:
                raise CircuitOpenError("Too many failures, circuit open")
            else:
                # Timeout 지나면 재시도 허용
                self.failures = 0

        try:
            result = func()
            self.failures = 0  # 성공 시 리셋
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = datetime.now()

            if self.failures >= self.threshold:
                # Circuit 열림 - 전체 배치 중단
                raise CircuitOpenError(
                    f"Circuit opened after {self.failures} failures"
                ) from e
            raise

# 사용
breaker = ConnectionCircuitBreaker(failure_threshold=5)

for task in tasks:
    try:
        result = breaker.execute(lambda: process_task(task))
    except CircuitOpenError:
        logger.error("Circuit opened, stopping batch")
        break  # 배치 중단하고 return

효과:

• DB 연결이 5번 연속 실패하면 즉시 배치 종료
• 무의미한 재시도로 시간 낭비 방지
• 함수가 빠르게 return되어 인스턴스 release

4. Graceful Degradation (청크 단위 처리)

전체를 한 번에 처리하지 않고 청크 단위로:

def run_probability_batch_chunked(
    grid_points: List[Tuple[float, float]] = None,
    chunk_size: int = 1000,
    **kwargs
):
    """청크 단위로 배치 처리"""

    if grid_points is None:
        grid_points = get_all_grid_points()

    # 격자점을 chunk_size 단위로 분할
    chunks = [
        grid_points[i:i+chunk_size]
        for i in range(0, len(grid_points), chunk_size)
    ]

    total_success = 0
    total_failed = 0

    for i, chunk in enumerate(chunks):
        logger.info(f"Processing chunk {i+1}/{len(chunks)}")

        try:
            # 청크당 timeout 설정
            with timeout_context(600):  # 10분
                run_probability_batch(
                    grid_points=chunk,
                    **kwargs
                )
            total_success += len(chunk)

        except TimeoutError:
            logger.error(f"Chunk {i} timeout, skipping")
            total_failed += len(chunk)
            continue  # 한 청크 실패해도 다음 청크 계속

        except Exception as e:
            logger.error(f"Chunk {i} failed: {e}")

            # 초반 3개 청크 실패 시 전체 중단
            if i < 3:
                raise

            total_failed += len(chunk)
            continue

    # 통계 로깅
    logger.info(f"Batch completed: {total_success} success, {total_failed} failed")

    # 항상 return 보장
    return {
        'success': total_success,
        'failed': total_failed
    }

효과:

• 한 청크가 실패해도 다른 청크는 처리됨
• 전체 배치 실패 위험 감소
• Progress tracking 용이

5. Dead Letter Queue (DLQ)

실패한 작업을 별도 저장하여 나중에 재처리:

def save_to_dead_letter_queue(task: Dict, error: str):
    """실패한 태스크를 DLQ에 저장"""
    db = DatabaseConnection()
    db.execute("""
        INSERT INTO batch_dead_letter_queue
        (task_type, task_data, error_message, created_at)
        VALUES (%s, %s, %s, NOW())
    """, ('probability', json.dumps(task), error))

# 배치 처리 중
for future in as_completed(futures):
    try:
        result = future.result(timeout=300)

        if result['status'] == 'failed':
            # DLQ에 저장
            save_to_dead_letter_queue(
                task=result['task'],
                error=result.get('error', 'Unknown error')
            )

    except TimeoutError:
        # Timeout도 DLQ에 저장
        save_to_dead_letter_queue(
            task=task,
            error='Task timeout after 300s'
        )

나중에 DLQ를 조회하여 실패한 태스크만 재처리:

def reprocess_dead_letter_queue():
    """DLQ의 실패한 태스크들을 재처리"""
    db = DatabaseConnection()
    failed_tasks = db.fetch_all(
        "SELECT * FROM batch_dead_letter_queue WHERE reprocessed = FALSE"
    )

    for record in failed_tasks:
        task = json.loads(record['task_data'])
        try:
            result = process_task(task)
            # 성공 시 DLQ에서 제거
            db.execute(
                "UPDATE batch_dead_letter_queue SET reprocessed = TRUE WHERE id = %s",
                (record['id'],)
            )
        except Exception as e:
            logger.error(f"Reprocess failed: {e}")

6. Health Check + 강제 종료

배치 상태를 외부에서 모니터링하여 강제 종료:

import redis
import os
import signal

# 배치 시작 시 상태 등록
def start_batch_monitoring(job_id: str, max_duration: int = 7200):
    """배치 시작을 Redis에 등록"""
    r = redis.Redis()
    r.hset(f'batch:{job_id}', mapping={
        'status': 'running',
        'pid': os.getpid(),
        'start_time': datetime.now().isoformat(),
        'max_duration': max_duration
    })

# 별도 모니터링 프로세스
def batch_monitor():
    """주기적으로 배치 상태 확인"""
    r = redis.Redis()

    while True:
        for key in r.scan_iter('batch:*'):
            info = r.hgetall(key)

            if info['status'] == 'running':
                start_time = datetime.fromisoformat(info['start_time'])
                running_time = (datetime.now() - start_time).seconds
                max_duration = int(info['max_duration'])

                if running_time > max_duration:
                    # 최대 실행 시간 초과 - 강제 종료
                    pid = int(info['pid'])
                    logger.warning(f"Killing hung batch process: PID {pid}")
                    os.kill(pid, signal.SIGTERM)

                    # 상태 업데이트
                    r.hset(key, 'status', 'killed')
                    r.hset(key, 'killed_at', datetime.now().isoformat())

        time.sleep(60)  # 1분마다 체크

7. APScheduler Job 설정 강화

# main.py
from apscheduler.executors.pool import ThreadPoolExecutor as APSThreadPoolExecutor

scheduler = BackgroundScheduler(
    executors={
        'default': APSThreadPoolExecutor(max_workers=2)
    },
    job_defaults={
        'coalesce': False,  # 밀린 작업 건너뛰기
        'max_instances': 1,  # 동시 실행 인스턴스 수
        'misfire_grace_time': 3600  # 1시간 이내 실행 실패 허용
    }
)

# Job 등록
scheduler.add_job(
    probability_batch_job,
    trigger=CronTrigger(month=1, day=1, hour=2, minute=0),
    id='probability_batch',
    name='P(H) Timeseries Batch',
    replace_existing=True,
    max_instances=1,
    misfire_grace_time=3600,  # 예정 시각 지나도 1시간 내 실행
    coalesce=True  # 밀린 작업 하나로 통합
)

실무 Best Practices

1. 계층별 Timeout 설정

┌─────────────────────────────────────────┐
│ APScheduler Job Level (3시간)        │
│  ┌───────────────────────────────────┐  │
│  │ Batch Function Level (2시간)    │  │
│  │  ┌─────────────────────────────┐  │  │
│  │  │ Chunk Level (10분)        │  │  │
│  │  │  ┌───────────────────────┐  │  │  │
│  │  │  │ Task Level (5분)    │  │  │  │
│  │  │  └───────────────────────┘  │  │  │
│  │  └─────────────────────────────┘  │  │
│  └───────────────────────────────────┘  │
└─────────────────────────────────────────┘

2. 실패 처리 전략

실패 유형별 처리:

1. Timeout (5분)
   → 해당 태스크만 실패 처리
   → DLQ에 저장
   → 다음 태스크 계속

2. DB Connection Error (Circuit Breaker)
   → 5번 연속 실패 시 Circuit Open
   → 배치 중단하고 return
   → 인스턴스 release

3. 데이터 오류 (개별 처리)
   → 로그 기록
   → 다음 태스크 계속

4. 심각한 에러 (시스템 레벨)
   → 배치 전체 중단
   → Exception raise
   → 알림 발송

3. 모니터링 지표

배치 실행 시 수집할 메트릭:

- start_time: 시작 시각
- end_time: 종료 시각
- duration: 실행 시간
- total_tasks: 전체 태스크 수
- completed_tasks: 완료된 태스크 수
- failed_tasks: 실패한 태스크 수
- timeout_tasks: Timeout된 태스크 수
- success_rate: 성공률
- avg_task_duration: 평균 태스크 처리 시간
- peak_memory: 최대 메모리 사용량
- db_connection_errors: DB 연결 에러 수

4. 프로덕션 체크리스트

• Task 레벨 timeout 설정 (5분)
• Batch 레벨 timeout 설정 (2시간)
• Circuit Breaker 구현
• Dead Letter Queue 구현
• Health Check 모니터링
• 실패 알림 설정 (Slack, Email)
• 메트릭 수집 및 대시보드
• 청크 단위 처리
• 재시도 메커니즘
• 로그 레벨 최적화

결론

근본 원인:

• ProcessPoolExecutor의 future.result()에 timeout이 없어서 Worker가 멈출 때 무한 대기
• 함수가 return되지 않아 APScheduler 인스턴스 카운트가 release되지 않음

핵심 해결책:

• future.result(timeout=300) 추가로 개별 태스크 timeout 설정
• Timeout 발생 시 해당 태스크만 실패 처리하고 배치 계속 진행
• 함수가 항상 return되도록 보장

추가 방어:

• Circuit Breaker로 연속 실패 시 빠른 종료
• 청크 단위 처리로 부분 실패 격리
• Dead Letter Queue로 실패한 작업 재처리
• Health Check로 외부 모니터링

효과:

• ✅ 배치 작업이 멈춰도 최대 5분 후 다음 작업 진행
• ✅ APScheduler 인스턴스가 항상 정상 release
• ✅ "maximum instances reached" 에러 해결
• ✅ 안정적인 배치 스케줄링 가능

관련 파일:

• modelops/batch/probability_timeseries_batch.py:271
• modelops/batch/hazard_timeseries_batch.py:297
• main.py:28-72

참고 자료:

• APScheduler Documentation
• Python concurrent.futures
• Circuit Breaker Pattern

사용자가 사업장 리스크 계산 API를 여러 번 호출했지만, 계산이 시작은 되나 완료되지 않고 멈춰버리는 문제가 발생했습니다.

로그 분석

2025-12-17 22:21:29,773 - modelops.batch.evaal_ondemand_api - INFO - Starting E, V, AAL calculation: (37.36633726, 127.10661717), SSP126, 2021
2025-12-17 22:21:29,774 - modelops.batch.evaal_ondemand_api - INFO - Starting E, V, AAL calculation: (37.36633726, 127.10661717), SSP126, 2022
2025-12-17 22:21:29,775 - modelops.batch.evaal_ondemand_api - INFO - Starting E, V, AAL calculation: (37.36633726, 127.10661717), SSP126, 2023
...
2025-12-17 22:21:30,824 - modelops.data_loaders.building_data_fetcher - WARNING - No 시도 found for sido_code=41

• 계산 시작 로그는 있지만 완료 로그가 없음
• 경고 메시지만 반복되고 실제 계산이 멈춤
• 에러 로그도 없이 조용히 멈춤

원인: Database Connection Pool 부재

기존 코드의 문제점

# modelops/database/connection.py (기존 코드)
class DatabaseConnection:
    """PostgreSQL 데이터베이스 연결 관리"""

    @staticmethod
    @contextmanager
    def get_connection():
        """데이터베이스 연결 컨텍스트 매니저"""
        conn = psycopg2.connect(  # ⚠️ 매번 새 연결 생성!
            DatabaseConnection.get_connection_string(),
            cursor_factory=RealDictCursor
        )
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            raise e
        finally:
            conn.close()

문제점:

• 매번 psycopg2.connect()를 호출하여 새로운 물리적 연결을 생성
• Connection Pool이 없어 연결 재사용 불가
• 동시 다발적인 연결 요청 시 PostgreSQL 서버에 과부하

왜 이것이 계산을 멈추게 했는가?

1. 병렬 처리 구조 분석

# modelops/api/routes/site_assessment.py
MAX_WORKERS = 8  # 8개의 Worker 스레드

def _background_calculate_site_risk(...):
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # 모든 작업 제출
        futures = []
        for site_id, site_location in sites.items():
            for scenario in SCENARIOS:  # 4개 시나리오
                for year in TARGET_YEARS:  # 80개 연도
                    future = executor.submit(
                        _calculate_single_site_scenario_year,
                        ...
                    )
                    futures.append(future)

병렬 처리 규모:

• 사업장 1개 × 시나리오 4개 × 연도 80개 = 320개 작업
• 8개 Worker 스레드가 동시 실행

2. 각 작업별 DB 연결 횟수

하나의 calculate_evaal_ondemand 호출 시 DB 연결이 발생하는 지점:

# 1. Hazard 조회 (9번 - 리스크 타입별)
fetch_hazard_from_db()
  → DatabaseConnection.fetch_hazard_results()
  → with DatabaseConnection.get_connection()  # 연결 1

# 2. Probability 조회 (9번)
fetch_probability_from_db()
  → DatabaseConnection.fetch_probability_results()
  → with DatabaseConnection.get_connection()  # 연결 2

# 3. 건물 정보 조회 (9번 - 각 리스크별로 HazardDataCollector 호출)
HazardDataCollector.collect_data()
  → building_fetcher.fetch_all_building_data()
  → get_building_code_from_coords()    # 연결 3
  → get_building_info()                # 연결 4
  → get_river_info()                   # 연결 5
  → get_distance_to_coast()            # 연결 6
  → get_population_data()              # 연결 7, 8, 9

# 4. DB 저장 (save_to_db=True인 경우)
_save_results_to_db()
  → save_exposure_results()        # 연결 10
  → save_vulnerability_results()   # 연결 11
  → save_aal_scaled_results()      # 연결 12

한 작업당 최소 12회 이상의 DB 연결 생성!

3. 동시 연결 요청 폭주

시점 T=0:
Thread 1: 작업 A 시작 → DB 연결 12개 생성
Thread 2: 작업 B 시작 → DB 연결 12개 생성
Thread 3: 작업 C 시작 → DB 연결 12개 생성
Thread 4: 작업 D 시작 → DB 연결 12개 생성
Thread 5: 작업 E 시작 → DB 연결 12개 생성
Thread 6: 작업 F 시작 → DB 연결 12개 생성
Thread 7: 작업 G 시작 → DB 연결 12개 생성
Thread 8: 작업 H 시작 → DB 연결 12개 생성

동시 연결 시도: 8 threads × 12 connections = 96개 연결!

4. PostgreSQL max_connections 한계 도달

PostgreSQL의 기본 max_connections 설정:

-- 일반적인 설정
max_connections = 100

문제 발생 시나리오:

1. 96개의 연결이 동시에 요청됨
2. PostgreSQL이 연결 생성 속도를 따라가지 못함
3. 일부 스레드는 연결을 기다리며 블로킹 상태로 진입
4. 연결 타임아웃이 발생하거나 데드락 상태에 빠짐
5. 예외가 스레드 내부에서 처리되어 메인 로그에 출력되지 않음

Thread 1: [======= 작업 중 =======]
Thread 2: [======= 작업 중 =======]
Thread 3: [==== 연결 대기 중... ====] ⏳
Thread 4: [==== 연결 대기 중... ====] ⏳
Thread 5: [==== 연결 대기 중... ====] ⏳
Thread 6: [==== 연결 대기 중... ====] ⏳
Thread 7: [==== 연결 대기 중... ====] ⏳
Thread 8: [==== 연결 대기 중... ====] ⏳
                 ↓
         계산이 멈춤!

해결: ThreadedConnectionPool 추가

수정된 코드

# modelops/database/connection.py (수정 후)
import psycopg2
from psycopg2 import pool  # ✅ 추가
from psycopg2.extras import RealDictCursor
from contextlib import contextmanager
from typing import List, Dict, Any, Optional
import uuid
import json
import logging
from datetime import datetime
from ..config.settings import settings

logger = logging.getLogger(__name__)


class DatabaseConnection:
    """PostgreSQL 데이터베이스 연결 관리"""

    # ✅ Connection Pool 추가 (스레드 안전)
    _connection_pool = None
    _pool_lock = None

    @classmethod
    def _init_pool(cls):
        """Connection Pool 초기화 (Lazy Initialization)"""
        import threading

        # 스레드 안전한 초기화를 위한 Lock
        if cls._pool_lock is None:
            cls._pool_lock = threading.Lock()

        with cls._pool_lock:
            if cls._connection_pool is None:
                try:
                    # ThreadedConnectionPool: 스레드 안전한 연결 풀
                    cls._connection_pool = pool.ThreadedConnectionPool(
                        minconn=2,   # 최소 유지 연결 수
                        maxconn=20,  # 최대 연결 수 (MAX_WORKERS=8 × 2.5 여유)
                        host=settings.database_host,
                        port=settings.database_port,
                        dbname=settings.database_name,
                        user=settings.database_user,
                        password=settings.database_password
                    )
                    logger.info("Database connection pool initialized (minconn=2, maxconn=20)")
                except Exception as e:
                    logger.error(f"Failed to initialize connection pool: {e}")
                    raise

    @staticmethod
    def get_connection_string() -> str:
        """데이터베이스 연결 문자열 생성"""
        return (
            f"host={settings.database_host} "
            f"port={settings.database_port} "
            f"dbname={settings.database_name} "
            f"user={settings.database_user} "
            f"password={settings.database_password}"
        )

    @classmethod
    @contextmanager
    def get_connection(cls):
        """데이터베이스 연결 컨텍스트 매니저 (Connection Pool 사용)"""
        # Pool 초기화 (처음 호출 시에만)
        if cls._connection_pool is None:
            cls._init_pool()

        conn = None
        try:
            # ✅ Pool에서 연결 가져오기 (기존 연결 재사용)
            conn = cls._connection_pool.getconn()
            conn.cursor_factory = RealDictCursor
            yield conn
            conn.commit()
        except Exception as e:
            if conn:
                conn.rollback()
            raise e
        finally:
            # ✅ Pool에 연결 반환 (close 대신 putconn)
            if conn:
                cls._connection_pool.putconn(conn)

주요 개선 사항

1. ThreadedConnectionPool 적용

pool.ThreadedConnectionPool(
    minconn=2,   # 항상 2개 연결 유지
    maxconn=20,  # 최대 20개까지 확장 가능
    ...
)

장점:

• 스레드 안전 (Thread-safe)
• 자동 연결 관리 (생성/재사용/회수)
• 내부적으로 Lock을 사용하여 동시성 제어

2. Lazy Initialization

if cls._connection_pool is None:
    cls._init_pool()

• 첫 호출 시에만 Pool 초기화
• 애플리케이션 시작 시 불필요한 연결 생성 방지
• 필요할 때만 리소스 사용

3. Double-Checked Locking

with cls._pool_lock:
    if cls._connection_pool is None:
        # Pool 생성

• 여러 스레드가 동시에 초기화를 시도해도 안전
• 한 번만 Pool이 생성되도록 보장

4. 연결 재사용

# 기존: 매번 새 연결
conn = psycopg2.connect(...)  # 느림 (TCP 핸드셰이크, 인증 등)
conn.close()                   # 연결 폐기

# 개선: Pool에서 재사용
conn = pool.getconn()          # 빠름 (기존 연결 재사용)
pool.putconn(conn)             # 반환 (연결 유지)

효과 비교

Before (Pool 없음)

시점 T=0:
Thread 1: psycopg2.connect() [300ms] → 작업 → close()
Thread 2: psycopg2.connect() [300ms] → 작업 → close()
Thread 3: psycopg2.connect() [300ms] → 작업 → close()
Thread 4: psycopg2.connect() [350ms] → 작업 → close()
Thread 5: psycopg2.connect() [400ms] → 작업 → close()
Thread 6: psycopg2.connect() [500ms] → 작업 → close()  ⚠️ 지연
Thread 7: psycopg2.connect() [1000ms] → 작업 → close() ⚠️ 큰 지연
Thread 8: psycopg2.connect() [타임아웃] ❌ 실패

문제점:
❌ 연결 생성 시간: 300~1000ms+ (누적)
❌ max_connections 한계 도달
❌ 타임아웃 및 실패 발생
❌ 메모리 낭비 (매번 새 연결)

After (Pool 적용)

시점 T=0:
Pool 초기화: 2개 연결 미리 생성 [600ms, 1회만]

Thread 1: pool.getconn() [5ms] → 작업 → putconn() ✅
Thread 2: pool.getconn() [5ms] → 작업 → putconn() ✅
Thread 3: pool.getconn() [10ms, 새 연결 생성] → 작업 → putconn() ✅
Thread 4: pool.getconn() [5ms, 재사용] → 작업 → putconn() ✅
Thread 5: pool.getconn() [5ms, 재사용] → 작업 → putconn() ✅
Thread 6: pool.getconn() [5ms, 재사용] → 작업 → putconn() ✅
Thread 7: pool.getconn() [5ms, 재사용] → 작업 → putconn() ✅
Thread 8: pool.getconn() [5ms, 재사용] → 작업 → putconn() ✅

개선 사항:
✅ 연결 획득 시간: 5~10ms (50~100배 빠름)
✅ 최대 20개 연결로 제한 (안정성)
✅ 연결 재사용 (메모리 효율)
✅ 타임아웃 없음
✅ 모든 스레드 정상 실행

성능 개선 지표

테스트 결과

Pool 적용 후 로그

2025-12-17 22:30:15,123 - modelops.database.connection - INFO - Database connection pool initialized (minconn=2, maxconn=20)
2025-12-17 22:30:15,456 - modelops.batch.evaal_ondemand_api - INFO - Starting E, V, AAL calculation: (37.366, 127.106), SSP126, 2021
2025-12-17 22:30:18,305 - modelops.batch.evaal_ondemand_api - INFO - E, V, AAL calculation completed: 2.85s  ✅
2025-12-17 22:30:18,310 - modelops.database.connection - INFO - Saved 9 exposure results  ✅
2025-12-17 22:30:18,315 - modelops.database.connection - INFO - Saved 9 vulnerability results  ✅
2025-12-17 22:30:18,320 - modelops.database.connection - INFO - Saved 9 AAL scaled results  ✅

결과:

• ✅ 계산이 정상 완료 (2.85초)
• ✅ DB에 성공적으로 저장
• ✅ Pool 초기화 로그 확인
• ✅ 더 이상 멈춤 현상 없음

핵심 포인트

왜 Connection Pool이 필수인가?

1. 연결 생성 비용이 매우 높음

   • TCP 3-way handshake
   • SSL/TLS 협상 (암호화 연결 시)
   • 사용자 인증
   • 세션 초기화
   • 총 300~1000ms 소요

2. 병렬 처리 환경에서 치명적

   • 여러 스레드가 동시에 연결 요청
   • PostgreSQL의 max_connections 한계
   • 연결 대기로 인한 성능 저하

3. 리소스 효율성

   • 연결 재사용으로 메모리 절약
   • DB 서버 부하 감소
   • 안정적인 처리량 보장

Connection Pool 설정 가이드

ThreadedConnectionPool(
    minconn=2,   # CPU 코어 수 정도
    maxconn=20,  # MAX_WORKERS × 2~3 정도
    ...
)

권장 설정:

• minconn: CPU 코어 수 또는 2~4 정도
• maxconn: Worker 스레드 수의 2~3배
• PostgreSQL max_connections: Pool의 maxconn × 여유율(1.5~2)

주의사항

# ❌ 잘못된 사용
conn = pool.getconn()
# 작업 수행
# putconn() 호출 안 함 → 연결 누수!

# ✅ 올바른 사용
conn = pool.getconn()
try:
    # 작업 수행
finally:
    pool.putconn(conn)  # 반드시 반환!

# ✅ 더 좋은 방법: Context Manager 사용
with DatabaseConnection.get_connection() as conn:
    # 작업 수행
    # 자동으로 putconn() 호출됨

결론

Connection Pool 부재가 병렬 처리 환경에서 계산을 멈추게 한 핵심 원인이었습니다. ThreadedConnectionPool을 적용하여 연결을 효율적으로 재사용하도록 수정한 결과, 계산이 정상적으로 완료되고 DB에 저장되는 것을 확인했습니다.

병렬 처리를 사용하는 환경에서 데이터베이스 연결은 반드시 Connection Pool을 통해 관리해야 하며, 특히 Python의 psycopg2에서는 ThreadedConnectionPool을 사용하여 스레드 안전성을 보장해야 합니다.

FastAPI에서 ThreadPoolExecutor 사용 시 앱 종료할 때 쓰레드 풀 정리 안 하면:

• 실행 중인 작업 강제 종료 → 데이터 손실
• 리소스 누수 (메모리, 파일 핸들)
• ResourceWarning: unclosed ThreadPoolExecutor 경고

해결

1. Service에 shutdown 메서드 추가

# src/services/report_service.py
from concurrent.futures import ThreadPoolExecutor

class ReportService:
    def __init__(self):
        self._executor = ThreadPoolExecutor(max_workers=4)

    def shutdown(self):
        """쓰레드 풀 정리"""
        if self._executor:
            self._executor.shutdown(wait=True)  # 실행 중인 작업 완료 대기
            self._executor = None

핵심: wait=True면 실행 중인 작업 완료까지 대기, False면 즉시 종료

2. FastAPI 이벤트 핸들러에서 호출

# main.py
from fastapi import FastAPI

app = FastAPI()

report_service_instance = None
analysis_service_instance = None

@app.on_event("startup")
async def startup_event():
    global report_service_instance, analysis_service_instance
    report_service_instance = ReportService()
    analysis_service_instance = AnalysisService()

@app.on_event("shutdown")
async def shutdown_event():
    global report_service_instance, analysis_service_instance

    if report_service_instance:
        report_service_instance.shutdown()

    if analysis_service_instance and hasattr(analysis_service_instance, 'shutdown'):
        analysis_service_instance.shutdown()

3. 백그라운드 작업은 데몬 스레드로

# ai_agent/utils/ttl_cleaner.py
import threading

def setup_background_cleanup(interval_hours: int = 1):
    def cleanup_loop():
        while True:
            cleanup_expired_sessions()
            time.sleep(interval_hours * 3600)

    # daemon=True: 메인 프로세스 종료 시 자동 종료
    thread = threading.Thread(target=cleanup_loop, daemon=True)
    thread.start()

데몬 스레드 특징:

• 메인 프로세스 종료하면 강제 종료됨
• 중요한 작업 X (데이터 손실 가능)
• 로그 정리, 캐시 정리 같은 부가 작업에만 사용

실행 흐름

앱 시작
  ↓
startup_event()
  ├─ ReportService 초기화 (ThreadPoolExecutor 생성)
  └─ Background cleanup 시작 (daemon thread)
  ↓
앱 실행
  ↓
Ctrl+C / SIGTERM
  ↓
shutdown_event()
  ├─ executor.shutdown(wait=True)  # 작업 완료 대기
  └─ daemon thread 자동 종료
  ↓
종료

비교

잘못된 방법

executor = ThreadPoolExecutor(max_workers=4)
# 앱 종료 시 executor 정리 안 함

올바른 방법

def shutdown(self):
    if self._executor:
        self._executor.shutdown(wait=True)
        self._executor = None

확인

앱 종료 시 로그 확인:

INFO: Application shutting down
INFO: Shutting down ReportService thread pool executor
INFO: ReportService shutdown complete
INFO: All services shut down successfully

ResourceWarning 경고 없으면 성공

추가 팁

Timeout 설정 (작업이 너무 오래 걸리는 경우):

def shutdown(self):
    if self._executor:
        # 최대 30초 대기
        self._executor.shutdown(wait=True)

Signal 직접 처리 (필요한 경우):

import signal
import sys

def signal_handler(sig, frame):
    shutdown_event()
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

백엔드 API를 배포하고 Swagger UI를 외부에서 접근 가능하게 하려다가 SSL 설정으로 한참을 헤맸다. Cloudflare와 Nginx Proxy Manager를 함께 쓰면서 발생한 여러 오류들과 해결 과정을 기록한다.

초기 구성

• Domain: example.site (가비아에서 구매, Cloudflare DNS 관리)
• 인프라: GCP VM + Docker
• 리버스 프록시: Nginx Proxy Manager
• 백엔드: Spring Boot (8080 포트)
• 목표: https://example.site/swagger-ui/index.html 접근 가능하게 만들기

시행착오 1: SSL handshake failed (Error 525)

초기 설정

- Cloudflare: 주황 구름 (프록시 활성화) + Full SSL 모드
- Nginx Proxy Manager: SSL Certificate None

결과

SSL handshake failed
Error code 525

원인 분석

• Cloudflare Full 모드는 Cloudflare ↔ 서버 간에도 HTTPS 통신을 요구한다
• 하지만 Nginx는 SSL 인증서가 없어서 HTTP만 제공
• 결과: Cloudflare가 HTTPS로 연결 시도 → Nginx가 응답 못함 → SSL handshake 실패

SSL 모드별 차이점

시행착오 2: Let's Encrypt HTTP Challenge 실패 (403)

처음에는 "Cloudflare 프록시(주황 구름)가 켜져 있으면 Let's Encrypt HTTP Challenge를 막을 거야"라고 생각했다.

예상한 문제

• Let's Encrypt는 /.well-known/acme-challenge/ 경로로 검증
• Cloudflare 프록시가 중간에서 이 요청을 차단할 것이다
• 따라서 회색 구름(DNS only)으로 바꿔야 한다

실제로는?

그냥 됐다. 😳

해결책: Cloudflare Full + Nginx Let's Encrypt

최종 작동 순서

graph TD
    A[Cloudflare Full + Nginx None] -->|525 에러| B[Nginx에서 Let's Encrypt 인증서 발급]
    B -->|HTTP Challenge 성공| C[Nginx에 인증서 적용]
    C -->|HTTPS 지원| D[접속 성공!]

구체적인 단계

1. 초기 상태 (에러 발생)

Cloudflare: Full 모드 + 주황 구름 (프록시)
Nginx: SSL None
결과: 525 에러

2. Let's Encrypt 인증서 발급

Nginx Proxy Manager 설정:

1. SSL Certificates → Add SSL Certificate
2. Let's Encrypt 선택
3. Domain Names: example.site
4. Email: 본인 이메일
5. Use a DNS Challenge: 체크 안 함 (HTTP Challenge)
6. I Agree to the Let's Encrypt Terms of Service 체크
7. Save 클릭

결과: 성공! 🎉

3. Proxy Host에 인증서 적용

1. Proxy Hosts에서 example.site 편집
2. SSL 탭:
   • SSL Certificate: 방금 생성한 Let's Encrypt 인증서 선택
   • Force SSL: 활성화
   • HTTP/2 Support: 활성화
3. Save

4. 접속 성공

https://example.site/swagger-ui/index.html → 정상 작동!

왜 이게 가능했을까?

Cloudflare의 특별한 ACME Challenge 처리

Cloudflare는 Full 모드에서도 Let's Encrypt ACME Challenge를 자동으로 감지하고 특별 처리한다.

일반 요청:
사용자 → Cloudflare (HTTPS) → Nginx (HTTPS 시도)
→ Nginx가 SSL 인증서 없으면 525 에러

ACME Challenge 요청:
Let's Encrypt → Cloudflare → Nginx (HTTP로 폴백)
→ Nginx가 HTTP로 토큰 응답
→ 인증서 발급 성공!

동작 원리

1. Let's Encrypt가 http://example.site/.well-known/acme-challenge/토큰 요청
2. Cloudflare가 이 경로를 감지
3. 백엔드에 HTTPS 연결 실패 → 자동으로 HTTP로 재시도
4. Nginx가 HTTP로 토큰 파일 응답
5. 인증서 발급 성공!

인증서 발급 후:

• Nginx가 HTTPS를 지원하게 됨
• Cloudflare Full 모드 정상 작동

핵심 요약

✅ 작동하는 방법

1. Cloudflare: Full 모드 + 주황 구름 유지
2. Nginx: SSL None 상태로 Proxy Host 생성
3. Nginx에서 Let's Encrypt HTTP Challenge로 인증서 발급
4. Proxy Host에 인증서 적용
5. 완료!

❌ 필요 없는 것들

• Cloudflare를 Flexible로 바꿀 필요 없음
• 회색 구름(DNS only)으로 바꿀 필요 없음
• DNS Challenge 사용할 필요 없음 (API 토큰 불필요)
• Cloudflare Origin Certificate 발급할 필요 없음

추가 팁

Nginx Proxy Manager 설정 확인사항

Proxy Host 설정:
  Domain Names: example.site
  Scheme: http
  Forward Hostname/IP: example-backend  # 컨테이너 이름
  Forward Port: 8080

  SSL 탭:
    SSL Certificate: Let's Encrypt 인증서 선택
    Force SSL: ON
    HTTP/2 Support: ON
    HSTS Enabled: ON (선택)

Docker 네트워크 확인

Nginx Proxy Manager와 백엔드 컨테이너가 같은 Docker 네트워크에 있어야 한다:

# 네트워크 생성
docker network create web

# 컨테이너 실행 시 네트워크 연결
docker run -d \
  --name example-backend \
  --network web \
  -p 8080:8080 \
  your-image

결론

Cloudflare는 개발자 친화적으로 설계되어 있어서, 프록시를 켠 상태로도 Let's Encrypt 인증서 발급이 가능하다.

처음에는 복잡하게 생각해서 여러 방법을 시도했지만, 결국 가장 간단한 방법이 정답이었다:

Cloudflare Full 모드 유지 → Nginx에서 Let's Encrypt HTTP Challenge → 인증서 적용

이것만 기억하면 된다! 🚀

참고 자료

• Cloudflare SSL/TLS 암호화 모드 문서
• Let's Encrypt ACME Challenge Types
• Nginx Proxy Manager 공식 문서

On-Do 팀은 기업의 사업장별 기후 물리 리스크를 예측하는 웹 서비스를 개발했다. Vue.js 프론트엔드, Spring Boot 백엔드, FastAPI AI 에이전트, 그리고 별도의 ModelOps 서버로 구성된 마이크로서비스 아키텍처를 채택했고, 이를 GCP(Google Cloud Platform)에 배포했다. 이 글에서는 왜 Kubernetes 대신 Docker + Nginx를 선택했는지, 그리고 Cloudflare, Nginx, Docker가 각각 어떤 역할을 하는지 실제 프로젝트 사례를 통해 설명하겠다.

전체 아키텍처

[사용자]
    ↓
[Cloudflare CDN + DDoS Protection]
    ↓
[DNS: on-do.site → GCP VM IP]
    ↓
[GCP VM Instance]
    ├─ [Nginx Proxy Manager] (:80, :443)
    │   ├─ on-do.site → polaris-frontend:80
    │   ├─ api.on-do.site → polaris-backend-java:8080
    │   └─ ai-agent-api.skax.co.kr → polaris-backend-fastapi:8000
    │
    ├─ [Docker Container: polaris-frontend] (Vue.js)
    ├─ [Docker Container: polaris-backend-java] (Spring Boot)
    ├─ [Docker Container: polaris-backend-fastapi] (FastAPI AI Agent)
    ├─ [별도 서버: ModelOps Server] (FastAPI ML Model Serving)
    └─ [PostgreSQL Database]

각 기술 스택의 역할

1️⃣ Cloudflare - CDN과 보안의 첫 번째 관문

역할: DNS 관리: on-do.site 도메인을 GCP 서버 IP로 연결 무료 SSL 인증서: Cloudflare Proxy를 통해 HTTPS 자동 적용 (프론트엔드) CDN: 정적 파일을 전 세계 엣지 서버에 캐싱하여 응답 속도 향상 DDoS 방어: 악의적인 트래픽 차단

우리 프로젝트에서: 프론트엔드(on-do.site)는 Cloudflare Proxy 사용 → CDN + 자동 HTTPS 백엔드(api.on-do.site)는 DNS only 모드 → Nginx에서 직접 SSL 처리 왜 백엔드는 Proxy를 끄나? Cloudflare Proxy를 켜면 Cloudflare가 중간에서 모든 요청을 처리하는데, 백엔드 API는 Nginx Proxy Manager에서 Let's Encrypt 인증서를 발급받아 직접 SSL을 처리하기 때문에 충돌이 발생한다.

2️⃣ Nginx Proxy Manager - 트래픽 라우팅의 중심

역할: 리버스 프록시: 도메인별로 요청을 적절한 Docker 컨테이너로 전달 SSL 인증서 관리: Let's Encrypt를 통해 무료 SSL 인증서 자동 발급 및 갱신 포트 매핑: 외부 80/443 포트를 내부 컨테이너의 다양한 포트로 연결 우리 프로젝트에서: on-do.site:443 → Nginx Proxy Manager → http://polaris-frontend:80

api.on-do.site:443 → Nginx Proxy Manager → http://polaris-backend-java:8080

ai-agent-api.skax.co.kr:443 → Nginx Proxy Manager → http://polaris-backend-fastapi:8000 설정 예시: Domain: on-do.site Scheme: http Forward Hostname: polaris-frontend Forward Port: 80 SSL Certificate: Let's Encrypt (자동 발급) Force SSL: ON 왜 Nginx Proxy Manager를 쓰나? GUI로 간편하게 설정 가능 (nginx.conf 직접 수정 불필요) SSL 인증서 자동 갱신 여러 서비스를 하나의 서버에서 도메인별로 분리

3️⃣ Docker - 컨테이너 기반 배포

역할: 각 서비스를 독립적인 컨테이너로 실행 환경 변수 주입으로 설정 관리 이미지 기반 배포로 재현성 보장 우리 프로젝트에서:

프론트엔드 컨테이너

docker run -d \
  --name polaris-frontend \
  --network web \
  -p 80:80 \
  asia-northeast3-docker.pkg.dev/.../polaris-frontend:latest

백엔드 컨테이너 (Spring Boot)

docker run -d \
  --name polaris-backend-java \
  --network web \
  -p 8080:8080 \
  -e SPRING_PROFILES_ACTIVE=prod \
  -e DB_HOST=10.117.192.3 \
  -e DB_PORT=5432 \
  -e JWT_SECRET=${{ secrets.JWT_SECRET }} \
  -e FASTAPI_API_KEY=${{ secrets.FASTAPI_API_KEY }} \
  asia-northeast3-docker.pkg.dev/.../polaris-backend-java:latest

AI Agent 서버 (FastAPI)

docker run -d \
  --name polaris-backend-fastapi \
  --network web \
  -p 8000:8000 \
  -e MODELOPS_API_URL=https://modelops.skax.co.kr \
  asia-northeast3-docker.pkg.dev/.../polaris-backend-fastapi:latest

Docker Network: 모든 컨테이너를 web 네트워크에 연결하여 컨테이너 이름으로 서로 통신: docker network create web 이렇게 하면: Spring Boot에서 http://polaris-backend-fastapi:8000로 AI Agent 호출 AI Agent에서 외부 ModelOps 서버로 ML 모델 추론 요청

4️⃣ 서비스 간 통신 구조

1. 사용자 요청 플로우:

[사용자] → "on-do.site에서 리스크 분석 요청"

[프론트엔드 Vue.js] → API 호출: POST https://api.on-do.site/api/analysis/start

[백엔드 Spring Boot] → 데이터베이스에 분석 작업 저장 → AI Agent 호출: POST http://polaris-backend-fastapi:8000/predict

[AI Agent FastAPI] → ModelOps 서버 호출: POST https://modelops.skax.co.kr/inference → ML 모델로 리스크 예측 → 결과를 Spring Boot로 반환

[백엔드 Spring Boot] → 결과를 데이터베이스에 저장 → 프론트엔드로 응답

[프론트엔드] → 사용자에게 리스크 점수 시각화

2. 내부 vs 외부 통신:

내부 (Docker 네트워크): 컨테이너 이름으로 통신 (빠름, 보안) polaris-backend-java → polaris-backend-fastapi 외부 (인터넷): 도메인으로 통신 polaris-backend-fastapi → modelops.skax.co.kr

5️⃣ 왜 Kubernetes를 쓰지 않았나?

Kubernetes의 장점:

자동 스케일링 (HPA: Horizontal Pod Autoscaler) 무중단 배포 (롤링 업데이트) 자가 치유 (Pod 장애 시 자동 재시작) 서비스 디스커버리, 로드 밸런싱 우리가 Kubernetes를 선택하지 않은 이유:

1. 프로젝트 규모가 작음

서비스 4개 (프론트엔드, 백엔드, AI Agent, ModelOps) 단일 VM 인스턴스에서 충분히 실행 가능 트래픽이 많지 않아 오토스케일링 불필요

2. 학습 곡선과 복잡도

Kubernetes는 다음을 모두 이해해야 함:

• Pod, Deployment, Service, Ingress
• ConfigMap, Secret
• kubectl 명령어
• YAML 설정 파일 작성
• Helm 차트 (패키지 관리)
• 클러스터 모니터링 반면 Docker + Nginx는: docker run -d --name myapp -p 8080:8080 myimage 이것만으로 배포 완료!

  3. 비용

  항목 Docker + Nginx Kubernetes (GKE) 컴퓨팅 e2-medium VM 1대
  $25/월 마스터 노드 + 워커 노드 3대
  $150~300/월 관리 비용 무료 (직접 관리) 클러스터 관리 비용 별도 네트워크 무료 (단일 VM) 로드 밸런서 비용 별도 우리의 선택: 3개월 프로젝트에 $300/월은 과하다!

  4. 배포 파이프라인이 간단함

  우리의 CI/CD (Docker + Nginx):

  GitHub Actions

1. 코드 푸시 (main 브랜치)
2. Docker 이미지 빌드
3. GCP Artifact Registry에 푸시
4. SSH로 서버 접속
5. docker pull && docker stop && docker run
6. Health Check (60초 대기) 만약 Kubernetes였다면:
7. 코드 푸시
8. Docker 이미지 빌드
9. Registry에 푸시
10. kubectl apply -f deployment.yaml
11. Ingress 설정 업데이트
12. 롤링 업데이트 모니터링
13. Pod 상태 확인
14. Service Mesh 설정 (Istio 등) 📊 실제 배포 플로우 CD 파이프라인 (GitHub Actions) name: CD - Deploy to Server

    on:
    workflow_run:
     workflows: ['CI - Build & Push']
     types: [completed]
     branches: [main]
    

jobs: deploy: runs-on: ubuntu-22.04 if: ${{ github.event.workflow_run.conclusion == 'success' }}

steps:
  - name: SSH로 서버 배포
    uses: appleboy/ssh-action@v1.2.0
    with:
      host: ${{ secrets.SERVER_HOST }}
      username: ${{ secrets.SERVER_USER }}
      key: ${{ secrets.SERVER_SSH_KEY }}
      script: |
        # GCP Artifact Registry 인증
        echo '${{ secrets.GCP_SA_KEY }}' | docker login -u _json_key --password-stdin https://asia-northeast3-docker.pkg.dev

        # 최신 이미지 pull
        docker pull asia-northeast3-docker.pkg.dev/.../polaris-backend-java:latest

        # 기존 컨테이너 중지 및 삭제
        docker stop polaris-backend-java || true
        docker rm polaris-backend-java || true

        # 새 컨테이너 실행
        docker run -d \
          --name polaris-backend-java \
          --network web \
          --restart unless-stopped \
          -p 8080:8080 \
          -e SPRING_PROFILES_ACTIVE=prod \
          -e JWT_SECRET="${{ secrets.JWT_SECRET }}" \
          -e DB_HOST="${{ secrets.DB_HOST }}" \
          -e DB_PORT="${{ secrets.DB_PORT }}" \
          -e DB_NAME="${{ secrets.DB_NAME }}" \
          -e DB_USERNAME="${{ secrets.DB_USERNAME }}" \
          -e DB_PASSWORD="${{ secrets.DB_PASSWORD }}" \
          -e SPRING_JPA_PROPERTIES_HIBERNATE_DIALECT=org.hibernate.dialect.PostgreSQLDialect \
          -e MAIL_HOST="${{ secrets.MAIL_HOST }}" \
          -e MAIL_PORT="${{ secrets.MAIL_PORT }}" \
          -e MAIL_USERNAME="${{ secrets.MAIL_USERNAME }}" \
          -e MAIL_PASSWORD="${{ secrets.MAIL_PASSWORD }}" \
          -e FASTAPI_API_KEY="${{ secrets.FASTAPI_API_KEY }}" \
          asia-northeast3-docker.pkg.dev/.../polaris-backend-java:latest

        # Health check (60초 타임아웃)
        echo "Health check 중..."
        for i in {1..60}; do
          if docker exec polaris-backend-java curl -f -s http://localhost:8080/actuator/health > /dev/null 2>&1; then
            echo "✓ 배포 성공! (${i}초 경과)"
            exit 0
          fi
          sleep 1
        done

        echo "✗ Health check 실패 (60초 타임아웃)"
        docker logs polaris-backend-java --tail 50
        exit 1

``` 배포 시 주의사항: 환경 변수 누락 방지: GitHub Secrets에 모든 필수 환경 변수 등록 Health Check 필수: 컨테이너가 정상 시작되었는지 확인 기존 컨테이너 정리: docker stop && docker rm 후 새 컨테이너 실행

우리 아키텍처의 장단점

장점

1. 단순성

배포 docker run -d --name myapp myimage

로그 확인 docker logs myapp

재시작 docker restart myapp 명령어 몇 줄로 모든 게 해결됨. Kubernetes의 복잡한 YAML 파일 불필요.

2. 비용 효율

단일 VM: e2-medium (2 vCPU, 4GB RAM) → $25/월 Kubernetes: GKE 클러스터 최소 구성 → $150/월 이상 3개월 프로젝트 기준: Docker + Nginx: $75 Kubernetes: $450 $375 절약!

3. 빠른 디버깅

컨테이너 내부 접속 docker exec -it polaris-backend-java bash

실시간 로그 docker logs -f polaris-backend-java

리소스 사용량 docker stats 문제 발생 시 바로 원인 파악 가능.

4. 충분한 성능

동시 접속자 수십~수백 명 처리 가능 Spring Boot의 Tomcat: 기본 200개 쓰레드 AI Agent의 Uvicorn: 비동기 처리로 높은 처리량

단점 (Kubernetes 대비)

1. 수동 스케일링

트래픽 급증 시 수동으로 인스턴스를 추가해야 함. Kubernetes는 자동 kubectl scale deployment myapp --replicas=10

우리는 수동 docker run -d --name myapp-2 myimage docker run -d --name myapp-3 myimage

2. 무중단 배포 불가

docker stop myapp # ← 이 순간 서비스 중단! docker run -d --name myapp myimage Kubernetes의 롤링 업데이트는 중단 없이 배포 가능. 해결 방법: Blue-Green 배포 Green (새 버전) 실행 docker run -d --name myapp-green myimage

Nginx 설정 변경: myapp → myapp-green

Blue (구 버전) 중지 docker stop myapp

3. 자가 치유 없음

컨테이너가 죽으면 수동으로 재시작해야 함. docker run -d --restart unless-stopped myapp # ← 이것으로 부분 해결 Kubernetes는 자동으로 Pod를 재시작함.

실제로 겪은 배포 문제들

1. 문제: MailConfig 때문에 ApplicationContext 로드 실패

증상: Failed to load ApplicationContext Caused by: Could not resolve placeholder 'spring.mail.host' 원인: MailConfig가 @Configuration으로 무조건 로드되는데, 메일 환경 변수가 없으면 실패. 해결: @Configuration @ConditionalOnProperty(name = "spring.mail.host") // ← 추가 public class MailConfig { // ... }

2. 문제: AWS S3 설정 때문에 시작 실패

증상: Could not resolve placeholder 'AWS_ACCESS_KEY' 원인: GCP로 전환했는데 AWS S3 설정이 남아있었음. 해결: S3Config.java 삭제 rm src/main/java/com/skax/physicalrisk/config/S3Config.java

application.yml에서 AWS 설정 제거

3. 문제: PostgreSQL 테이블이 없어서 실패

증상: Schema-validation: missing table [analysis_jobs] 원인: ddl-auto: validate로 설정되어 테이블이 없으면 실패. 해결: application-prod.yml jpa: hibernate: ddl-auto: update # validate → update 변경

4. 문제: Docker 컨테이너끼리 통신 안 됨

증상: curl: (6) Could not resolve host: polaris-backend-fastapi 원인: 컨테이너들이 다른 네트워크에 있었음. 해결: 공통 네트워크 생성 docker network create web

모든 컨테이너를 web 네트워크에 연결 docker run -d --network web --name polaris-backend-java ... docker run -d --network web --name polaris-backend-fastapi ...

언제 Kubernetes로 전환해야 할까?

다음 상황이 오면 Kubernetes를 고려해야 함:

1. 트래픽 폭증

동시 접속자: 수백 명 → 수천~수만 명 → 오토스케일링 필수

2. 마이크로서비스 확장

서비스 개수: 4개 → 10개 이상 → 서비스 디스커버리, 로드 밸런싱 필요

3. 고가용성 요구

SLA: 99% → 99.9% 이상 → 멀티 리전 배포, 자동 장애 복구 필요

4. 멀티 클라우드

GCP + AWS + Azure 동시 사용 → Kubernetes는 클라우드 벤더 중립적

정리

항목 | Docker + Nginx | Kubernetes 학습 곡선 낮음 ⭐ 높음 ⭐⭐⭐⭐⭐ 초기 비용 낮음 ($25/월) 높음 ($150/월) 배포 복잡도 낮음 높음 확장성 제한적 무제한 무중단 배포 수동 (Blue-Green) 자동 (Rolling) 자가 치유 제한적 (restart 옵션) 자동 (Pod 재시작) 모니터링 수동 (docker stats) 자동 (Prometheus) 적합한 규모 소규모중규모 중규모~대규모

우리의 선택: Docker + Nginx ✅ 3개월 프로젝트 기간 ✅ 4개 서비스 (프론트엔드, 백엔드, AI Agent, ModelOps) ✅ 중소규모 트래픽 (동시 접속자 수십~수백 명) ✅ 제한된 예산 ($75 vs $450) ✅ 빠른 배포와 디버깅

배운 점

1. 적정 기술 선택의 중요성

"최신 기술 = 좋은 기술" 이 아니다. 프로젝트 규모와 요구사항에 맞는 기술을 선택하는 게 중요하다.

2. 인프라는 단순할수록 좋다

복잡한 인프라는 디버깅도 어렵고, 팀원 온보딩도 어렵다. Docker + Nginx는 누구나 이해할 수 있다.

3. 비용 최적화

스타트업이나 소규모 프로젝트에서는 비용이 중요하다. Kubernetes로 $375를 절약한 건 큰 성과다.

4. 확장 가능한 설계

지금은 Docker + Nginx지만, 나중에 Kubernetes로 전환할 수 있도록 설계했다: 12 Factor App 원칙 준수 환경 변수로 설정 관리 컨테이너 기반 배포 상태를 저장하지 않는 Stateless 서비스

"기술은 목적이 아니라 수단이다. 과하지도 부족하지도 않은, 딱 맞는 기술을 선택하자."

백엔드 구조를 정리할 때 가장 먼저 등장하는 레이어 개념이 있다.

Route(Controller) / Service / Schema(DTO) / Core(Domain)

각 레이어의 역할만 정확히 알고 있어도 코드 구조가 훨씬 깔끔해진다.

1. Route Layer (Controller, Router)

HTTP 요청을 받고, 어떤 서비스 함수로 보낼지 “매핑”하는 레이어.

하는 일

• URL + HTTP 메서드 정의 (GET /users/{id})

• 요청 데이터를 Schema(DTO)로 변환

• 예외 처리 → HTTP 응답 포맷 작성

• Service 호출 → 결과 받아서 반환

하지 말아야 하는 일

• 비즈니스 로직

• DB 접근

• 트랜잭션 처리

Route는 얇게, Service는 두껍게가 유지보수에 유리하다.

2. Service Layer (Application Service)

“요청 단위의 유스케이스”를 실행하는 레이어.

하는 일

• 여러 도메인 객체(Core)를 조합해 기능 수행

• Repository(DB) 접근을 순서대로 오케스트레이션

• 트랜잭션 처리

• 권한/검증 등의 애플리케이션 규칙 처리

예시

회원가입 → 중복검사 → 비밀번호 해싱 → DB 저장

주문 생성 → 재고 차감 → 결제 요청 → 기록 저장

Service 레이어가 전체 플로우를 총괄 지휘한다.

3. Schema Layer (DTO, Pydantic Model)

API 입력/출력 구조 + 유효성 검증을 담당.

하는 일

• 요청/응답의 필드 정의

• 타입 검증, 형식 검증

• 도메인 모델과 분리된 “I/O 전용” 데이터 모델

Schema는 “외부 세계와 내부 도메인 사이의 방화벽 역할”을 한다. API 스펙이 바뀌어도 Core (도메인)을 바로 건드리지 않아도 된다.

4. Core Layer (Domain)

비즈니스 핵심 규칙을 포함하는 레이어.

해야하는 일

• Entity, Value Object (User, Product, Money 등)

• 순수 비즈니스 로직 (할인 계산, 위험 점수 계산 등)

• Repository 인터페이스 (구현체는 infra에서)

Core는 웹 프레임워크나 DB 종류에 의존하면 안된다. 이렇게 해야 FastAPI → Spring Boot로 갈아타도 Core를 그대로 재사용 가능하다.

전체 구조 요약

레이어    역할
Route    HTTP 라우팅, 요청/응답 구성
Service    유스케이스 실행, 비즈니스 흐름 조립
Schema    요청·응답 포맷 관리, 유효성 검증
Core    순수 도메인 규칙, 핵심 모델

Spring Boot vs FastAPI — 기술적인 차이

1. 언어/생태계

Spring Boot

Java/Kotlin 기반

대규모 엔터프라이즈 백엔드 표준

보안/트랜잭션/배치/메시징 등 인프라 스택 완비

FastAPI

Python 기반

AI/데이터 생태계와 결합이 매우 쉬움

가볍고 개발 속도가 빠름

2. 프레임워크 성격

Spring Boot = Full-stack 백엔드 프레임워크 기업 서비스 전체 운영을 위한 종합 프레임워크

FastAPI = Lightweight 웹 프레임워크 빠른 개발, 확장성 확보, ML·AI 모델 API화에 적합

3. IO/비동기 처리 방식

Spring Boot

기본은 동기식

WebFlux(Reactor)가 필요 시 고성능 비동기 처리 가능

장기간 운영 시 JVM 최적화가 강력

FastAPI

애초에 async/await 기반

IO-bound API에 유리 (ML inference, 외부 API 호출 등)

ESG 물리 기후 예측 프로젝트는 왜 Spring Boot + FastAPI를 둘 다 썼는가?

1. 전체 구조가 “AI 모델 운영(ModelOps) + 백엔드 서비스”로 분리된 구조였기 때문

해당 프로젝트는 단순 CRUD API가 아니라 다음과 같이 2계층 구조였다:

백엔드 서비스 계층 (Spring Boot)

회원관리

인증(JWT)

채팅 로그 저장

안정적인 API 스펙 제공

트래픽 대응

DB 트랜잭션 및 보안 담당

AI/모델 연산 계층 (FastAPI, Python)

ChatGPT API 연동

Pre/Post-processing

간단한 LLM 라우팅

분석/ML 로직과 Python 생태계 필요

Python의 강점은 다음과 같음:

NLP, AI 라이브러리(PyTorch, Transformers 등) 빠른 개발 속도 ML 추론/전처리에 최적화

반면 Spring Boot는 다음이 강함:

서비스 운영 안정성 인증/보안 구조화된 레이어드 아키텍처 대규모 트래픽 처리

그래서 각 기술의 장점을 그대로 사용한 것임.

2. 프론트(Android/React Native)에서 호출하는 메인 API는 Spring이 더 적합했기 때문

_Spring은 다음이 강력하다: _

엄격한 타입 검증 명확한 패키지 구조 스케일링 및 운영 안정성 기업용 API 스펙에 익숙한 구조

반면 FastAPI는:

AI inference API 실험용/내부용 API 비동기 요청 처리 이런 용도로 더 잘 맞는다.

→ 즉, 서비스 API와 모델 API를 분리해서 의존성을 줄이는 구조로 운영한 것.

3. 단일 프레임워크로 모든 것을 처리하려고 하면 오히려 손해였기 때문

FastAPI로 회원가입/로그인/JWT/DB 관리까지 하려면 결국:

ORM 설정
인증 체계
계층 구조
배포 파이프라인

등을 전부 별도로 만들어야 한다.

Spring Boot는 이미 이 기능셋이 정교하게 갖춰져 있으므로 백엔드 메인 API는 Spring을 쓰는 것이 더 효율적이었다.

반대로, Spring Boot로 AI inference API를 만들면, Python 기반 라이브러리를 쓰기 어려워지고 LLM 처리 속도, 유연성, 개발 효율이 크게 떨어진다.

마무리

Route–Service–Schema–Core 구조는 “역할을 명확하게 분리해 유지보수를 쉽게 하는 방법”이다. 이 구조를 적용하면 코드가 복잡해질수록 더 깔끔해지고, 팀 개발에서도 충돌이 줄어든다.

또한, Spring Boot와 FastAPI를 동시에 사용한 이유는 단순히 기술 욕심이 아니라, 각 프레임워크가 잘하는 영역이 명확히 달랐기 때문이다.

Spring Boot → 인증, 트랜잭션, 안정적인 서비스 운영
FastAPI → AI 모델 연산, Python 기반 전처리/후처리, 비동기 IO

즉, 해당 프로젝트는 “하나로 다 해결하는 프레임워크”를 고른 것이 아니라, 역할에 따라 옳은 도구를 선택한 것이다.

두 기술을 조합하면 서비스 안정성과 AI 처리 성능을 동시에 확보할 수 있으며, 앞으로 기능을 확장할 때도 각 계층을 독립적으로 개선할 수 있다.

결국 중요한 것은 기술 스택이 아니라 구조화된 설계와 역할 분리다.

FastAPI로 AI Agent를 호출하는 API를 개발하던 중, 성능 문제를 발견했습니다.

# ❌ 문제가 있는 코드
async def create_report(self, request: CreateReportRequest) -> dict:
    analyzer = self._get_analyzer()

    # 동기 함수를 그냥 호출 → 이벤트 루프 블로킹!
    result = analyzer.analyze(
        target_location,
        building_info,
        asset_info,
        analysis_params
    )

    return result

문제점

1. 이벤트 루프 블로킹: analyzer.analyze()는 동기 함수인데, async 함수 내에서 직접 호출
2. 동시성 상실: 다른 요청들이 현재 분석이 끝날 때까지 대기해야 함
3. 응답 시간 증가: 분석에 30초 걸리면 다른 요청도 30초 이상 대기

왜 문제가 될까?

FastAPI는 비동기 이벤트 루프를 사용합니다:

[이벤트 루프]
  ├─ 요청 A 처리 (async)
  ├─ 요청 B 처리 (async)
  └─ 요청 C 처리 (async)

하지만 동기 함수를 직접 호출하면:

[이벤트 루프]
  ├─ 요청 A 처리 중... (30초 동기 작업 실행 중)
  │   ↓ 다른 요청들 모두 대기...
  │   ↓
  │   ↓ (30초 경과)
  ├─ 요청 B 처리 시작 (뒤늦게)
  └─ 요청 C 처리 시작 (더 늦게)

💡 해결 방법: ThreadPoolExecutor

동기 함수를 별도 스레드에서 실행하여 이벤트 루프를 블로킹하지 않도록 개선합니다.

1. 필요한 모듈 임포트

import asyncio
from concurrent.futures import ThreadPoolExecutor
from functools import partial

2. Service 클래스에 ThreadPool 추가

class ReportService:
    def __init__(self):
        self._analyzer = None
        self._report_results = {}
        # 최대 4개의 worker 스레드로 ThreadPool 생성
        self._executor = ThreadPoolExecutor(max_workers=4)

max_workers=4의 의미:

• 최대 4개의 분석 작업을 동시에 처리 가능
• CPU 코어 수와 메모리를 고려하여 조정 (권장: CPU 코어 수 × 2)

3. 비동기 실행 패턴 적용

async def create_report(self, request: CreateReportRequest) -> dict:
    analyzer = self._get_analyzer()

    # Language 파라미터 준비
    language = request.language.value if request.language else 'ko'

    # ✅ 개선된 코드: 비동기 실행
    loop = asyncio.get_event_loop()

    # partial로 함수와 인자를 미리 바인딩
    analyze_func = partial(
        analyzer.analyze,
        target_location,
        building_info,
        asset_info,
        analysis_params,
        language=language
    )

    # ThreadPool에서 실행 (이벤트 루프는 블로킹되지 않음!)
    result = await loop.run_in_executor(self._executor, analyze_func)

    return result

🔧 핵심 개념 설명

1. asyncio.get_event_loop()

현재 실행 중인 이벤트 루프 인스턴스를 가져옵니다.

loop = asyncio.get_event_loop()

2. functools.partial()

함수와 인자를 미리 결합하여 새로운 함수를 만듭니다.

# 원본 함수
def analyze(location, building, asset, params, language):
    ...

# partial로 인자를 미리 바인딩
analyze_func = partial(
    analyze,
    location_data,
    building_data,
    asset_data,
    params_data,
    language='ko'
)

# 나중에 인자 없이 호출 가능
result = analyze_func()  # 위에서 바인딩한 인자들이 자동으로 전달됨

왜 필요한가?

run_in_executor()는 인자가 없는 callable을 받기 때문에, partial로 인자를 미리 묶어둬야 합니다.

# ❌ 이렇게는 안 됨
result = await loop.run_in_executor(
    executor,
    analyzer.analyze(location, building, ...)  # 즉시 실행되어 버림!
)

# ✅ partial로 감싸면 됨
analyze_func = partial(analyzer.analyze, location, building, ...)
result = await loop.run_in_executor(executor, analyze_func)

3. loop.run_in_executor(executor, func)

ThreadPool의 별도 스레드에서 함수를 실행하고, 완료될 때까지 await로 대기합니다.

result = await loop.run_in_executor(self._executor, analyze_func)

동작 원리:

[Main Thread - Event Loop]
  ├─ 요청 A 시작
  ├─ run_in_executor() 호출 → Worker Thread 1에 작업 전달
  ├─ 요청 B 시작 (블로킹 안 됨!)
  ├─ run_in_executor() 호출 → Worker Thread 2에 작업 전달
  └─ 요청 C 시작 (블로킹 안 됨!)

[Worker Thread 1]
  └─ analyzer.analyze() 실행 중... (30초)

[Worker Thread 2]
  └─ analyzer.analyze() 실행 중... (30초)

📊 성능 비교

Before (동기 호출)

요청 1: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초)
요청 2:                               ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초)
요청 3:                                                               ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초)

총 소요 시간: 90초

After (비동기 실행)

요청 1: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초)
요청 2: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초) ← 동시 실행!
요청 3: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ (30초) ← 동시 실행!

총 소요 시간: 30초

3배 빠른 처리! (max_workers=4 기준, 4개까지 동시 처리 가능)

🎯 실전 적용 팁

1. Worker 수 설정

import os

# CPU 코어 수에 따라 동적 설정
max_workers = min(32, (os.cpu_count() or 1) * 2)
self._executor = ThreadPoolExecutor(max_workers=max_workers)

2. Executor 정리 (Graceful Shutdown)

class ReportService:
    def __init__(self):
        self._executor = ThreadPoolExecutor(max_workers=4)

    def __del__(self):
        """서비스 종료 시 ThreadPool 정리"""
        self._executor.shutdown(wait=True)

3. 타임아웃 설정

import asyncio

try:
    # 60초 타임아웃
    result = await asyncio.wait_for(
        loop.run_in_executor(self._executor, analyze_func),
        timeout=60.0
    )
except asyncio.TimeoutError:
    return {"error": "Analysis timeout"}

🚨 주의사항

1. GIL (Global Interpreter Lock)

Python의 GIL로 인해 CPU-bound 작업은 ThreadPool로 성능 개선이 제한적입니다.

• I/O-bound 작업: ThreadPool 효과 큼 (네트워크 요청, 파일 읽기 등)
• CPU-bound 작업: ProcessPoolExecutor 사용 권장 (CPU 집약적 계산)

우리의 경우는 LLM API 호출이 포함되어 I/O-bound이므로 ThreadPool이 효과적입니다.

2. 상태 공유 주의

여러 스레드에서 동시에 접근하는 변수는 Thread-safe해야 합니다.

# ❌ 위험: 여러 스레드에서 동시 수정
self._report_results[report_id] = result

# ✅ 안전: Lock 사용
import threading

class ReportService:
    def __init__(self):
        self._lock = threading.Lock()
        self._report_results = {}

    async def create_report(self, request):
        result = await loop.run_in_executor(...)

        with self._lock:
            self._report_results[report_id] = result

📝 전체 코드

from concurrent.futures import ThreadPoolExecutor
from functools import partial
import asyncio

class ReportService:
    def __init__(self):
        self._analyzer = None
        self._report_results = {}
        self._executor = ThreadPoolExecutor(max_workers=4)

    def _get_analyzer(self):
        if self._analyzer is None:
            from ai_agent import SKAXPhysicalRiskAnalyzer
            from ai_agent.config.settings import load_config

            config = load_config()
            self._analyzer = SKAXPhysicalRiskAnalyzer(config)
        return self._analyzer

    async def create_report(self, request: CreateReportRequest) -> dict:
        analyzer = self._get_analyzer()

        # 데이터 준비
        target_location = {...}
        building_info = {...}
        asset_info = {...}
        analysis_params = {...}
        language = request.language.value if request.language else 'ko'

        # 비동기 실행
        loop = asyncio.get_event_loop()
        analyze_func = partial(
            analyzer.analyze,
            target_location,
            building_info,
            asset_info,
            analysis_params,
            language=language
        )
        result = await loop.run_in_executor(self._executor, analyze_func)

        return result

    def __del__(self):
        self._executor.shutdown(wait=True)

🎓 핵심 요약

핵심 패턴:

loop = asyncio.get_event_loop()
func = partial(sync_function, arg1, arg2, ...)
result = await loop.run_in_executor(executor, func)

이 패턴을 사용하면 FastAPI에서 동기 라이브러리를 사용하면서도 비동기의 장점을 그대로 누릴 수 있습니다! 🚀

📚 참고 자료

• asyncio — Asynchronous I/O
• concurrent.futures — Launching parallel tasks
• FastAPI - Async / Await

대규모 계산 작업(E, V, AAL 계산 등)과 일반 API 요청이 섞여 있는 시스템에서는 비동기 처리를 도입하는 순간 동기 방식에서는 절대 겪지 않는 새로운 문제들이 튀어나온다.

아래에서는 실제로 ModelOps·AI Agent 구조에서 공통으로 부딪히는 문제들을 나누어 설명하고, 각 문제를 어떻게 설계 레벨에서 해결해야 하는지 정리한다.

1. 고전적 문제: API 요청과 Heavy 계산이 섞이면서 생기는 병목

비동기라 하더라도, 계산과 API 요청이 같은 프로세스 / 같은 워커에서 돌아가면 계산이 CPU를 오래 점유하며 API 응답이 밀리는 문제가 발생한다.

E,V,AAL 계산이 8초 걸리는 동안 연속으로 들어오는 /health or /status API조차 응답이 늦어짐

해결 전략

1) 계산 서버(Calc Service)와 API 서버를 분리한다

API 서버는 "job 생성"과 "결과 조회"만 담당 계산은 별도 FastAPI(혹은 워커)에서 수행

2) Compute 서버는 반드시 별도 쓰레드풀/워커 구조로 운영

FastAPI BackgroundTask Celery worker RQ worker multiprocessing 기반 worker

→ API 이벤트 루프와 계산 이벤트 루프를 절대로 공유하지 않는다.

2. 중복 트리거 문제 (Double Trigger)

“같은 사이트/같은 시나리오/같은 연도”에 대해 중복으로 계산 요청(job)이 들어오는 문제.

해결 전략

1) (site_id, scenario, horizon) 조합에 Unique 제약 추가

UNIQUE(site_id, scenario, horizon)

이러면 중복 요청이 들어오면:

기존 job 리턴 or "이미 계산 중" status만 줘도 됨

2) "Idempotency Token" 패턴 도입

API 호출 시 idempotency_key를 받고 같은 키면 같은 job 반환.

3. 비동기 처리에서 가장 흔한 문제: “이 계산이 끝났다는 것을 어떻게 아는가?”

LLM/Business 서버는 비동기 계산이 얼마나 걸릴지 모르고, ModelOps는 계산 중인데, 프론트/Agent는 결과가 필요한 상황.

해결 전략

1) Job 테이블을 도입

job_id
  status = PENDING | RUNNING | COMPLETED | FAILED
  error_message
  started_at
  finished_at

Agent는 다음 API만 사용:

POST /jobs → job 생성 GET /jobs/{id} → 상태 조회 GET /sites/{id}/results → 결과 조회

Agent는 절대 계산 결과를 ModelOps API에서 직접 받지 않는다.

4. 작업 실패 처리 난이도 증가제

동기 시스템은 실패하면 API가 500을 그냥 내보내버린다. 하지만 비동기 시스템에서는:

워커가 실패했는지 실패했으면 재시도할 건지 실패 로그가 어딨는지 Agent가 실패를 사용자에게 어떻게 전달할지

관리 포인트가 늘어난다.

해결 전략

1) Job 상태를 FAILED로 기록 2) error_message/log_url 저장 3) Agent는 status를 기준으로 리포트 처리

FAILED면 “이번 분석은 실패했습니다. 나중에 다시 시도하세요.” COMPLETED면 결과 조회 후 LLM 리포트 생성

5. Race Condition (경쟁 조건)

A와 B 두 요청이 같은 job에 접근해서 다음과 같은 문제가 생김:

A가 status를 RUNNING으로 변경 B도 거의 동시에 status를 RUNNING으로 변경 → 중복 계산 발생 → 결과가 엉킴

해결 전략

1) DB level Lock 사용 (SELECT FOR UPDATE) 2) Job state transition 로직을 원자적으로 설계 3) Unique 제약과 Double-start 방지 로직 강화

6. 작업 폭주 문제: 동시에 너무 많은 작업이 들어오는 상황

10명만 써도 동시에 “후보지 8개”씩 계산하면 ModelOps에 80개 job이 쌓임 → CPU 폭주 → 전체 시스템 지연.

해결 전략

1) Worker concurrency 제한

Celery → --concurrency=N
ThreadPool → max_workers 설정

2) Job 큐 길이 제한

Job을 넣기 전에 검증:

if current_running_jobs > MAX:
    return 429 Too Many Requests

3) Job 우선순위(Priority Queue)

사업장 분석 > 후보지 분석 최근 요청 > 오래된 요청

7. 결과 저장 전략의 혼란

비동기 시스템은 결과를 어디에 저장할 것인지도 난제다.

잘못된 패턴: 결과를 계산 API 리턴으로 주려고 함 → 비동기와 완전히 충돌

해결 전략

E, V, AAL 결과는 반드시 DB(Site-level)로 저장해야 한다.

site_results
--------------
site_id
scenario
horizon
e_score
v_score
aal_value
risk_index
updated_at

Agent는 항상 DB만 읽는다.

정리: 안전한 아키텍처 패턴

    [User]
        ↓
   [AI Agent API]
        ↓  (트리거)
POST /jobs
        ↓
   [ModelOps API]
        ↓  (백그라운드/워커)
   Compute Worker
        ↓
     [DB]
        ↑
[AI Agent] ← GET /jobs/{id}, GET /results

핵심 요약

API와 계산이 섞임 -> 서버 분리 / 워커 사용 중복 계산 -> Unique key / idempotency 상태 추적 -> Job 테이블 도입 실패 -> FAILED 상태 관리 경쟁 조건 -> Lock 사용 작업 폭주 -> concurrency 제한 결과 저장 위치 ->DB 고정

마무리

비동기 작업 시스템은 성능 때문에 도입하지만, 성능보다 더 어려운 상태 관리와 일관성 문제가 새롭게 등장한다.

위의 문제와 해결책을 미리 고려하면 ModelOps–AI Agent–Backend가 섞인 복잡한 시스템에서도 안정적으로 “계산 요청 → 비동기 처리 → 결과 조회 → 보고서 생성” 전체 생산라인을 손대지 않고 확장할 수 있다.

• TCFD 공시용 ESG 보고서에 기재될 기후 물리적 리스크에 대한 점수, 연평균자산손실률 그리고 보고서 작성

대용량 시계열 데이터 처리

“대용량 시계열 데이터는 DB를 반복 조회하는 대신, 한 번만 로드해서 Staging 영역에 저장하고, 파이프라인 내부에서는 Scratch Space(임시 디스크)에서 재사용하며, TTL 기반 자동 정리로 디스크 부하를 방지하는 구조로 설계”