이 프로젝트에서 가장 많이 배운 건 코드를 짤 때가 아니라 뭔가 터졌을 때였다. 4편은 운영 중에 실제로 발생한 사고들을 시간 순서 없이, 기억나는 것들 위주로 정리한 글이다.

사고 1. KDS 재생성 후 Slack 알림이 30분 동안 안 왔다

비용 절감을 위해 인프라를 내렸다가 다시 올리는 작업을 했다. KDS를 삭제하고 재생성하면 Firehose 연결이 자동으로 붙을 줄 알았다.

아니었다.

Firehose는 KDS가 삭제되는 순간 SUSPENDED 상태로 전환되고, 이후 새 KDS가 생겨도 자동으로 reconnect되지 않는다. 직접 다시 연결해줘야 한다. 이걸 몰라서 Bronze 적재가 0 rows인 채로 한참 돌았다.

Lambda도 같은 문제가 있었다. Alert KDS에 연결된 Lambda event source mapping이 Disabled 상태로 고착된다.

# Disabled 항목 확인
aws lambda list-event-source-mappings \
    --event-source-arn <alert-kds-arn>

# 재활성화
aws lambda update-event-source-mapping \
    --uuid <mapping-uuid> \
    --enabled

이걸 발견한 건 Flink가 Alert KDS에 17건을 보냈는데 Slack 알림이 하나도 안 왔을 때였다. 30분 동안 운영자 화면에서 아무 알림도 없었던 것. Lambda가 KDS를 읽지 않고 있었다.

지금은 down.sh / up.sh 스크립트에 이 재활성화 스텝을 명시적으로 포함했다. 까먹을 수 없게.

사고 2. Grafana에서 로봇이 1,000대 중 480대만 보였다

가장 오래 헤맨 사고다. 진단에만 1시간 넘게 걸렸다.

Grafana Anomaly Timeline 패널에서 COUNT(DISTINCT robot_id)를 5분 슬라이딩 윈도우로 집계하고 있었다. 어느 날 값이 갑자기 절반 수준으로 떨어졌다. 480, 490, 510... 1,000 근처가 아니라 딱 절반이었다.

처음엔 Generator가 절반만 데이터를 보내는 줄 알았다. CloudWatch KDS IncomingRecords를 보니 정상이었다.

그 다음엔 Athena 쿼리 문제인 줄 알았다. 직접 돌려보니 정상이었다.

실제 원인은 Firehose의 multi-shard flush stagger였다.

KDS 2개 Shard가 있을 때, 각 Shard의 Firehose flush 타이밍이 약 15초씩 어긋난다. Shard 1이 12:00:00에 flush하고, Shard 2가 12:00:15에 flush한다. PartitionKey(robot_id) 해시 기준으로 Shard가 나뉘기 때문에, 5분 슬라이딩 윈도우 안에 두 Shard의 데이터가 완전히 겹치지 않는 타이밍이 생긴다.

결국 5분 윈도우 안에 Shard 1 로봇만 잡히는 순간 → COUNT(DISTINCT robot_id) ≈ 500. 딱 절반이 나왔던 이유가 MD5 해시 분포가 Shard 경계와 거의 정확히 50:50으로 나뉘었기 때문이었다.

해결은 간단했다. 윈도우를 Firehose buffer 시간 + 여유 마진 이상으로 늘렸다.

Firehose buffer: 300초 (5분)
Shard 수: 2
→ 최소 윈도우: 300 + (2 × 60) = 420초 (7분)
→ 실제 적용: 10분 윈도우

근데 이걸 파악하기 전까지 "Generator가 일부 로봇을 샘플링하는 게 아닐까?", "S3 파티션이 일부만 쓰이는 건 아닐까?" 등 전혀 다른 방향을 한참 들여다봤다. 결정적인 단서는 항상 딱 절반이 나온다는 거였다.

사고 3. Slack Webhook URL이 코드에 박혔다

초반에 빠르게 테스트하려고 Slack Webhook URL을 코드에 직접 넣었다. GitHub에 올라갔다. Public repo였다.

Slack이 자동으로 탐지해서 Webhook을 즉시 폐기했다. 다행히 Slack 쪽에서 먼저 막아줬지만 찜찜했다.

이후 모든 민감값은 AWS Secrets Manager에 저장하고 Lambda에서 런타임 조회로만 쓴다. Terraform에서도 TF_VAR_slack_webhook_url 환경변수로 주입하던 방식을 버리고 Secrets Manager에서 직접 read하는 방식으로 바꿨다.

data "aws_secretsmanager_secret_version" "slack_webhook" {
  secret_id = "/robot-telemetry/slack-webhook-url"
}

resource "aws_lambda_function" "alert_handler" {
  environment {
    variables = {
      SLACK_WEBHOOK_URL = data.aws_secretsmanager_secret_version.slack_webhook.secret_string
    }
  }
}

환경변수로 주입하면 export를 한 번만 빼먹어도 CHANGEME 같은 기본값이 prod에 조용히 들어간다. Secrets Manager에서 직접 read하면 이 패턴 자체가 없어진다.

사고 4. ALB를 안 지우고 셧다운했다

비용 절감을 위해 EKS 워크로드를 전부 0으로 줄이는 셧다운 루틴을 만들었다. Pod을 전부 0으로 줄이면 노드가 없어지고, Karpenter가 EC2를 반납한다.

근데 ALB가 살아있었다.

Pod을 0으로 줄여도 kubectl delete ingress를 안 하면 ALB Controller가 ALB를 그대로 유지한다. AWS가 2024년 2월부터 public IPv4 주소를 attached/unattached 무관하게 과금하기 시작했다.

ALB 3개 × public IP 2개씩 × $0.005/h × 720h = 월 $21.6 누수.

셧다운 루틴에 Ingress 삭제 + ALB 완전 삭제 확인을 추가했다.

# Ingress 삭제
kubectl delete ingress --all -A

# ALB가 완전히 사라질 때까지 대기
until [ "$(aws elbv2 describe-load-balancers \
    --query 'length(LoadBalancers)' --output text)" = "0" ]; do
    echo "ALB 삭제 대기 중..."
    sleep 30
done
echo "ALB 완전 삭제 확인"

Pod만 0으로 줄이고 ALB가 살아있으면 노드는 사라지는데 public IP는 계속 과금되는 상황이 된다. 셧다운 루틴은 ALB까지 확인하고 끝내야 한다.

사고 5. dt = D-1 하드코딩

API 서버에서 캐시를 갱신할 때 이렇게 썼었다.

yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
query = f"SELECT * FROM gold WHERE dt = '{yesterday}'"

평소엔 괜찮다. 근데 비용 절감을 위해 인프라를 내린 날 밤에 Airflow ETL이 돌지 않으면? gold 테이블에 어제 파티션이 없다. 캐시 갱신 시 빈 DataFrame이 올라가고, AI 채팅은 "데이터를 찾을 수 없습니다"만 반환하게 된다. 자연 회복이 안 된다.

다음날 인프라를 다시 올려도, 오늘 ETL이 돌기 전까지 어제 파티션은 없는 채로 유지된다.

# 수정: 최근 N일 안에서 MAX(dt) 사용
query = """
    SELECT * FROM gold
    WHERE dt = (
        SELECT MAX(dt) FROM gold
        WHERE dt >= DATE_FORMAT(DATE_ADD(NOW(), INTERVAL -7 DAY), '%Y-%m-%d')
    )
"""

7일 안에서 가장 최신 파티션을 쓴다. ETL이 하루 빠져도 이틀 전 데이터라도 보여준다. 빈 응답보다 낫다.

사고 6. SageMaker SDK 기본 버킷 문제

Airflow에서 SageMaker 학습 job을 실행할 때 이런 에러가 났다.

HeadBucket Error: An error occurred (403) when calling the HeadBucket operation
Bucket: sagemaker-eu-west-1-123456789012

SageMaker SDK가 자동으로 sagemaker-{region}-{account} 버킷을 기본 버킷으로 쓰려 하는데, Airflow의 IRSA에는 그 버킷 접근 권한이 없었다.

# ❌ 기본 버킷 사용 (IRSA 권한 없음)
sagemaker.Session()

# ✅ 명시적 버킷 지정
sagemaker.Session(default_bucket="robot-telemetry-bucket")

이것 말고도 source_dir 상대경로 문제도 있었다. Airflow worker pod의 현재 디렉토리가 /opt/airflow라서, DAG 파일 기준 상대경로가 맞지 않았다.

# ❌ 상대경로 (worker pod CWD 기준)
source_dir="src/ml"

# ✅ 절대경로 (__file__ 기준 파생)
source_dir=str(Path(__file__).parent.parent / "src" / "ml")

SageMaker 관련 문제가 많았던 이유 중 하나가 Airflow worker pod에 SageMaker SDK를 전역으로 설치했던 거다. SageMaker SDK가 313MB라서, webserver/scheduler pod이 시작될 때마다 pip install이 돌았다. 콜드스타트 6분 → CrashLoop.

worker pod에만 격리해서 설치하는 방식으로 바꿨다.

# helm/airflow-values.yaml
workers:
  env:
    - name: _PIP_ADDITIONAL_REQUIREMENTS
      value: "sagemaker boto3"  # worker만

# webserver/scheduler에는 해당 없음

비용 최적화 — 월 $750 절감

이 프로젝트를 운영하면서 비용을 꽤 줄였다.

KDS Shard를 4개에서 2개로 줄인 건, 실제 트래픽을 측정해보니 1,000 rec/sec가 항상 나오는 게 아니었기 때문이다. 피크를 커버할 수 있는 최소한으로 줄였다. (나중에 1개까지 줄이는 실험도 했는데, 그게 위에 나온 Firehose flush stagger 사고로 이어졌다.)

Partition Projection이 가장 효과가 컸다. DDL에 파티션 범위를 수식으로 정의해두면 Athena가 S3 메타데이터 조회 없이 스캔할 파티션을 바로 계산한다.

TBLPROPERTIES (
  'projection.enabled' = 'true',
  'projection.dt.type' = 'date',
  'projection.dt.range' = '2026-01-01,NOW',
  'projection.dt.format' = 'yyyy-MM-dd',
  'projection.dt.interval' = '1',
  'projection.dt.interval.unit' = 'DAYS'
)

이 프로젝트가 해커톤으로 이어진 것들

이 파이프라인을 만들면서 직접 써본 것들이 PRISM 해커톤에서 그대로 살아났다.

• Bedrock 연동 — 여기서 처음 써봤다. 해커톤에서 4 Agent + Supervisor 구조를 빠르게 잡을 수 있었던 건 이미 한 번 해봤기 때문이다.
• XGBoost 예측 모델 — AI4I 2020 데이터셋으로 여기서 처음 학습시켰다. PRISM의 6-class 분류기의 베이스였다.
• 결정론적 시연 — 이 프로젝트에서 KDS 재생성 후 Lambda 고착 같은 사고를 경험하면서 "시연 환경은 절대 외부 의존성을 최소화해야 한다"는 걸 배웠다. PRISM의 LLM 캐시 replay 구조가 그 결과물이다.
• 운영 가드레일 — 여기서 터진 사고들이 그대로 CLAUDE.md 가드레일 19개가 됐다.

학습 프로젝트로 시작했지만, 결국 해커톤 MVP의 기술적 토대가 됐다. 처음부터 대규모를 목표로 설계하지 않고, 작게 만들고 부딪히면서 키워나가는 방식이 맞았다고 생각한다.

Flink가 실시간을 담당한다면, Airflow는 하루치 데이터를 정제하고 쌓는 역할이다. 그리고 그 위에 Bedrock AI 채팅과 SageMaker 예측 모델이 올라간다.

Medallion Architecture — Bronze, Silver, Gold

S3 Data Lake를 세 계층으로 나눴다. 처음엔 "굳이 이렇게까지 해야 하나?" 싶었는데, 운영하다 보니 이 구조가 없으면 나중에 엄청 골치 아파진다는 걸 알게 됐다.

Bronze (원본)
  s3://bucket/bronze/year=2026/month=04/day=27/hour=12/
  → Firehose가 그대로 적재, 절대 수정 안 함
  → 90일 후 S3 Glacier로 이동

    ↓ Athena ETL (매일 자정)

Silver (정제)
  s3://bucket/silver/dt=2026-04-27/
  → 이상치 제거, 중복 제거, null 처리
  → 365일 보관

    ↓ Athena ETL (일 1회)

Gold (집계)
  s3://bucket/gold/dt=2026-04-27/
  → robot_id별 일일 집계값
  → 영구 보관 (분석 자산)

Bronze를 절대 수정하지 않는 게 핵심이다. Silver/Gold 로직에 버그가 있어도 Bronze에서 다시 재처리하면 된다. 원본이 오염되면 복구가 불가능하다.

Airflow — 멱등성이 가장 중요한 설계 원칙

Airflow DAG가 매일 자정에 돌면서 Bronze → Silver → Gold → Bedrock Report를 처리한다.

dag = DAG(
    'robot_daily_etl',
    schedule_interval='0 0 * * *',
    catchup=False
)

quality_check >> bronze_to_silver >> silver_to_gold >> bedrock_report

Airflow에서 가장 중요하게 생각했던 건 멱등성이었다. 같은 DAG를 두 번 실행해도 결과가 동일해야 한다.

초기에 이 쿼리를 썼다.

-- ❌ 비멱등 (재실행하면 데이터 중복)
INSERT INTO TABLE silver_robot_telemetry
PARTITION (dt = '2026-04-27')
SELECT ...

재실행하면 같은 파티션에 데이터가 두 배로 쌓인다. 이걸 발견한 건 테스트 중에 DAG를 두 번 돌렸다가 집계값이 이상하게 나왔을 때였다.

-- ✅ 멱등 (재실행해도 동일)
INSERT OVERWRITE TABLE silver_robot_telemetry
PARTITION (dt = '2026-04-27')
SELECT ...

OVERWRITE를 쓰면 기존 파티션을 지우고 새로 쓴다. 몇 번을 실행해도 결과가 같다.

Bronze → Silver 정제

INSERT OVERWRITE TABLE silver_robot_telemetry
PARTITION (dt = '{{ ds }}')
SELECT
    robot_id, pos_x, pos_y, battery_level,
    motor_temp, current_load, timestamp
FROM bronze_robot_telemetry
WHERE year = YEAR(CAST('{{ ds }}' AS DATE))
  AND month = MONTH(CAST('{{ ds }}' AS DATE))
  AND day = DAY(CAST('{{ ds }}' AS DATE))
  AND motor_temp < 500          -- 이상치 제거
  AND battery_level BETWEEN 0 AND 100
  AND robot_id IS NOT NULL
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY robot_id, timestamp ORDER BY robot_id
) = 1                           -- 중복 제거

실제로 Bronze에는 같은 robot_id + timestamp 조합이 두 번 들어오는 경우가 있었다. Generator가 KDS 전송 실패 후 재시도할 때 생기는 중복이다. ROW_NUMBER()로 처음 것만 남긴다.

Silver → Gold 집계

INSERT OVERWRITE TABLE gold_robot_daily_stats
PARTITION (dt = '{{ ds }}')
SELECT
    '{{ ds }}'                              AS dt,
    robot_id,
    AVG(motor_temp)                         AS avg_motor_temp,
    MAX(motor_temp)                         AS max_motor_temp,
    100 - MIN(battery_level)               AS battery_drain,
    COUNT(DISTINCT HOUR(timestamp))         AS active_hours
FROM silver_robot_telemetry
WHERE dt = '{{ ds }}'
GROUP BY dt, robot_id

1,000대 × 86,400초 = 8,640만 레코드가 1,000행으로 압축된다. 이 Gold 테이블이 API 캐시와 AI 분석의 원천이다.

Great Expectations — 데이터 품질 게이트

Bronze 데이터에 문제가 있으면 Silver/Gold까지 오염된다. ETL 시작 전에 검증 게이트를 뒀다.

검증 규칙:
- robot_id null 비율 < 1%
- motor_temp: 0 ~ 500°C 범위
- battery_level: 0 ~ 100% 범위
- 레코드 수 > 0

검증 실패하면 DAG가 중단되고 Slack으로 알림이 간다. 처음에 이걸 귀찮아서 뺐다가, Generator 버그로 motor_temp = 999 데이터가 들어온 걸 Gold까지 올라간 뒤에야 발견했다. 그 이후로 절대 빼지 않는다.

FastAPI Portal + In-Memory Cache

API 서버는 운영자가 쓰는 포털의 백엔드다. 가장 많이 쓰이는 기능이 AI 채팅이었다.

"ROBOT-00042의 상태를 분석해줘" 같은 질문을 보내면, 해당 로봇의 Gold 데이터를 가져와서 Bedrock Claude에게 전달하고 분석을 받아온다.

처음엔 채팅 요청이 올 때마다 Athena 쿼리를 날렸다.

Athena 쿼리 흐름:
1. 쿼리 시작: ~1초
2. S3 스캔: 5~10초
3. 결과 직렬화: ~1초
→ 총 응답 시간: 10초 이상

10초는 채팅 UX에서 너무 길다. Gold 데이터는 하루에 한 번만 갱신되는데, 매번 Athena를 쓰는 건 비효율적이었다.

In-Memory Cache로 바꿨다.

_gold_cache: dict = {}      # {robot_id: {...stats}}
_cache_ready: bool = False

def refresh_cache():
    """매일 01:00 KST에 Gold 데이터를 메모리에 로드"""
    global _gold_cache, _cache_ready
    _cache_ready = False

    df = athena_query(f"SELECT * FROM gold WHERE dt = '{yesterday}'")
    _gold_cache = df.set_index('robot_id').to_dict('index')
    _cache_ready = True

# APScheduler로 매일 자동 갱신
scheduler.add_job(
    refresh_cache, 'cron',
    hour=1, minute=0,
    timezone='Asia/Seoul'  # ← 타임존 명시 필수
)

Gold 데이터 1,000행 × 6컬럼이 메모리에 올라가도 약 1MB다. 이걸 Dict로 들고 있으면 채팅 요청 때 캐시 조회가 1ms다.

타임존을 Asia/Seoul로 명시한 건 실수 한 번 하고 나서다. 처음엔 타임존 없이 hour=1로만 설정했더니 UTC 01:00(= KST 10:00)에 갱신되고 있었다. 운영자가 아침에 출근해서 "왜 어제 데이터가 아직도 나와?"라고 물어봤을 때 발견했다.

Bedrock AI 채팅

@app.post("/api/chat")
async def chat(question: str):
    if not _cache_ready:
        raise HTTPException(503, "캐시 로드 중입니다")

    # 캐시에서 즉시 조회
    robot_data = _gold_cache.get(robot_id_from_question(question))

    # Bedrock Claude 호출
    response = bedrock.invoke_model(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        body=json.dumps({
            "system": "당신은 공장 로봇 정비팀의 기술 고문입니다. 로봇 ID는 반드시 [ROBOT-XXXXX] 형식으로 표기하세요.",
            "messages": [{
                "role": "user",
                "content": f"데이터: {robot_data}\n질문: {question}"
            }],
            "max_tokens": 512
        })
    )

    answer = parse_response(response)

    # [ROBOT-XXXXX] 패턴 감지 → 딥링크 자동 생성
    links = extract_robot_links(answer)

    return {"response": answer, "links": links}

응답에서 [ROBOT-00042] 같은 패턴을 정규식으로 찾아내면, 해당 로봇의 Grafana 대시보드로 이동하는 딥링크 버튼을 자동으로 만든다. 운영자가 AI 답변을 보다가 특정 로봇이 언급되면 버튼 클릭 하나로 Grafana로 넘어간다.

XSS 방지를 위해 AI 응답을 HTML로 렌더링할 때 DOMPurify로 sanitize한다.

SageMaker XGBoost — 예측 정비

실시간 이상탐지는 지금 이상한 거 잡는 거고, SageMaker 모델은 미래에 고장날 것 같은 로봇을 미리 찾는 게 목적이다.

입력: [avg_motor_temp, max_motor_temp, battery_drain, active_hours]
출력: 고장 확률 (0~1), 위험 등급 (low/medium/high)

Gold 테이블 30일치 데이터를 학습 데이터로 쓰고, AI4I 2020 데이터셋의 machine_failure 레이블을 정답으로 사용한다. 주간 자동 재학습으로 모델이 최신 데이터를 반영하도록 했다.

# dags/weekly_ml_retrain.py (매주 월요일)
if execution_date.weekday() == 0:
    train_model(lookback_days=30)
    # 이전 모델보다 성능 높으면 배포, 아니면 유지

여기서 꽤 골치 아픈 문제들이 있었는데 — SageMaker SDK 버전, IRSA 권한, Airflow worker pod 설정 등 — 전부 4편에서 정리한다.

Grafana 대시보드

Grafana는 Athena와 CloudWatch를 데이터소스로 연결해서 세 가지 대시보드를 운영했다.

1. Fleet Status — 1,000대 전체 상태 (온도, 배터리, 가동 시간)
2. Anomaly Timeline — 이상 이벤트 발생 패턴 시계열
3. Pipeline Health — KDS throughput, Firehose 적재율, Airflow 성공률

Anomaly Timeline 패널에서 처음에 짧은 시간 윈도우(5분)를 썼다가 데이터가 이상하게 보여서 한참 헤맸던 사고가 있다. Firehose의 multi-shard flush stagger 때문이었는데, 이것도 4편에서 자세히 다룬다. 진단에만 1시간 넘게 썼다.

4편에서는 이 프로젝트를 운영하면서 실제로 터진 사고들을 정리한다. 코드보다 운영에서 배운 게 훨씬 많았다.

센서 데이터 1개가 Slack Alert이 되기까지 걸린 시간은 약 6초였다. 이 편에서는 Generator부터 Flink 이상탐지까지, 데이터가 실제로 흘러가는 경로를 따라간다.

Generator — 1,000개 코루틴

Generator는 가상 로봇 1,000대를 시뮬레이션하는 Python 프로세스다. K8s Deployment로 배포되고, 각 로봇마다 독립적인 asyncio 코루틴이 돌아간다.

async def robot_coroutine(robot_id: str):
    while True:
        record = {
            "robot_id": robot_id,
            "motor_temp": generate_temp(robot_id),
            "battery_level": generate_battery(robot_id),
            "current_load": generate_load(robot_id),
            "timestamp": datetime.utcnow().isoformat()
        }
        await send_to_kds(record)
        await asyncio.sleep(1)  # 1초마다

# 1,000개 코루틴 동시 실행
tasks = [robot_coroutine(f"ROBOT-{i:05d}") for i in range(1000)]
await asyncio.gather(*tasks)

초당 1,000건이 KDS로 나간다. 실제로는 500건씩 묶어서 put_records 배치 호출을 한다. 500 × 2회/초 = 1,000 rec/sec.

여기서 중요한 설계 결정이 하나 있다. PartitionKey를 robot_id로 설정했다.

kinesis_client.put_records(
    StreamName="robot-telemetry-stream",
    Records=[{
        "Data": json.dumps(record),
        "PartitionKey": record["robot_id"]  # ← 핵심
    }]
)

같은 robot_id는 항상 같은 Shard로 라우팅된다. 덕분에 Flink에서 robot_id별 시계열 처리를 할 때 순서가 보장된다. 로봇 A의 데이터가 뒤섞여서 들어오면 5분 이동평균 계산이 틀어진다.

Glue Schema Registry

KDS로 데이터를 보내기 전에 Glue Schema Registry로 스키마 검증을 한다. 필드 타입이 맞는지, 필수 필드가 빠지지 않았는지 확인한다.

처음엔 "어차피 우리가 만드는 데이터인데 검증이 왜 필요해?"라고 생각했다. 실제로 Generator 코드를 수정하다가 필드명을 motorTemp에서 motor_temp로 바꾼 적이 있었다. Schema Registry가 없었다면 Flink에서 조용히 null 처리되고 이상탐지가 통째로 멈췄을 거다.

Kinesis Firehose — Bronze 적재

KDS에 들어온 데이터는 두 군데로 나간다.

1. Managed Flink — 실시간 이상탐지 (수백 ms)
2. Kinesis Firehose — S3 Bronze 저장 (5분 배치)

Firehose 설정에서 중요한 두 가지.

Parquet 변환. JSON을 그대로 저장하면 S3 비용과 Athena 스캔 비용이 모두 올라간다. Firehose가 자동으로 JSON → Parquet(Snappy) 변환을 해준다. 실제로 같은 데이터 기준으로 JSON 대비 약 4분의 1 크기로 줄었다.

Dynamic Partitioning. S3 경로를 자동으로 나눠준다.

s3://bucket/bronze/
  year=2026/month=04/day=27/hour=12/
    <UUID>.parquet
  year=2026/month=04/day=27/hour=13/
    <UUID>.parquet

나중에 Athena에서 쿼리할 때 WHERE hour = 12 조건을 주면 hour=12 폴더만 스캔한다. 이게 없으면 하루치 데이터 전부를 읽어야 한다.

Managed Flink — 실시간 이상탐지

Flink 애플리케이션이 KDS에서 데이터를 읽어 두 가지 조건으로 이상을 탐지한다.

조건 1: Z-Score 기반

단순히 motor_temp > 90°C 같은 고정 임계값을 쓰지 않은 이유가 있다.

로봇마다 정상 동작 온도가 다르다. 어떤 로봇은 평소에 80°C에서 안정적으로 돌고, 다른 로봇은 60°C가 기준이다. 80°C 로봇에게 "90°C 넘으면 이상"이라고 하면 진짜 이상과 정상 변동을 구분하기 어렵다.

그래서 robot_id별 5분 이동평균과 표준편차를 계산하고, 현재 값이 얼마나 벗어났는지를 Z-Score로 본다.

-- Flink SQL: 5분 OVER Window로 통계 계산
SELECT
    robot_id,
    event_time,
    motor_temp,
    AVG(motor_temp) OVER (
        PARTITION BY robot_id
        ORDER BY event_time
        RANGE INTERVAL '5' MINUTE PRECEDING
    ) AS avg_temp,
    STDDEV(motor_temp) OVER (...) AS stddev_temp
FROM source_kds

Z = (현재값 - 5분평균) / 표준편차

ROBOT-00042 예시:
- 5분 평균: 75°C, 표준편차: 4.2°C
- 현재: 90°C
- Z = (90 - 75) / 4.2 = 3.57 → 이상 (Z > 3.0)

GREATEST(stddev_temp, 0.5) 가드를 넣어둔 건, 표준편차가 0일 때(모든 값이 동일할 때) 나눗셈이 무한대가 되는 걸 막기 위해서다.

조건 2: 다변량 상관성 (부하 대비 온도)

Z-Score만으로는 잡지 못하는 케이스가 있다. 고부하 작업 중에 온도가 올라가는 건 정상이다. 그런데 부하가 낮은데 온도가 높은 건 모터 문제다.

WHERE motor_temp >= 85.0
  AND (motor_temp / GREATEST(current_load, 1.0)) > 1.8

두 조건 중 하나라도 참이면 이상으로 판정한다.

Watermark — Late Data 처리

Event Time 기반 Window를 쓸 때 반드시 해결해야 하는 문제가 있다. Window가 언제 닫히는가?

네트워크 지연으로 데이터가 늦게 도착할 수 있다. 12:05 Window가 닫혔는데 12:04:55 타임스탬프를 가진 데이터가 12:05:12에 도착하면 어떻게 해야 하는가?

WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND

"10초 이상 늦은 데이터는 버린다"고 선언하면, Window는 window_end + 10초 이후에 확정된다. 이게 없으면 Window가 영원히 열려 있어 메모리가 계속 쌓인다.

로봇 센서 데이터는 거의 실시간으로 오기 때문에 10초 허용이 적당했다.

알람 집계 + 이중 Sink

이상이 탐지되면 1분 Tumbling Window로 한 번 더 집계한다. 같은 로봇이 연속으로 이상 신호를 보내도 1분에 1번만 Alert가 나가게 하기 위해서다.

# Statement Set으로 두 Sink를 동일 트랜잭션에 실행
statement_set = t_env.create_statement_set()
statement_set.add_insert_into("sink_alert_kds", alert_events)  # Lambda 트리거
statement_set.add_insert_into("sink_s3_alerts", alert_events)  # 이력 보존
statement_set.execute()

한쪽 Sink가 실패하면 둘 다 롤백된다. Exactly-Once 보장.

Lambda → Slack

Alert KDS에 데이터가 들어오면 Lambda가 자동으로 트리거된다.

def lambda_handler(event, context):
    for record in event['Records']:
        payload = json.loads(base64.b64decode(record['kinesis']['data']))

        # SSM에서 포털 URL 런타임 조회 (배포 시점에 모름)
        portal_url = ssm.get_parameter(
            Name='/robot-telemetry/portal-url'
        )['Parameter']['Value']

        slack_message = f"""
⚠️ *이상 감지*
🤖 {payload['robot_id']}
🌡️ {payload['max_alert_temp']:.1f}°C
🔗 <{portal_url}/?robot_id={payload['robot_id']}|포털에서 확인>
        """

        sns.publish(TopicArn=SNS_TOPIC_ARN, Message=slack_message)

포털 URL을 코드에 하드코딩하지 않은 이유: ALB DNS는 EKS 배포 후에야 확정된다. 배포 시점에 모르는 값은 SSM Parameter Store에 런타임 조회로 처리하는 게 깔끔하다.

Slack Webhook URL도 .env가 아니라 AWS Secrets Manager에 저장해서 Lambda가 런타임에 읽어온다. 이걸 코드에 하드코딩했다가 한 번 사고가 났었는데, 그 이야기는 4편에서 한다.

전체 타임라인

총 6초. 센서에서 운영자 알림까지.

3편에서는 배치 ETL과 AI 서빙 레이어를 다룬다. Airflow로 Bronze → Gold를 만들고, Bedrock으로 AI 채팅을 붙이는 과정이다.

이전 글에서 데이콘 스마트 제조 AI 해커톤 PRISM을 소개했는데, 사실 그 프로젝트의 뼈대는 몇 달 전부터 혼자 만들던 미니 프로젝트에서 왔다. PRISM에서 Bedrock AI 연동, XGBoost 예측 모델, Kinesis 스트리밍을 비교적 빠르게 붙일 수 있었던 건 이미 이 프로젝트에서 한 번씩 다 겪어봤기 때문이다.

시작하게 된 이유

데이터 엔지니어링을 공부하면서 "실제로 데이터가 흘러가는 시스템"을 한 번이라도 직접 만들어보고 싶었다. 강의에서 배운 Kafka, Spark 같은 개념들이 막상 AWS 환경에서 어떻게 연결되는지가 잘 그려지지 않았다.

그래서 선택한 소재가 가상 로봇 fleet의 센서 데이터였다. AI4I 2020 생산 설비 고장 데이터셋(CC BY 4.0)을 기반으로, 로봇 수천 대가 실시간으로 센서 데이터를 보내는 상황을 시뮬레이션하는 거다.

처음엔 "로봇 100대 정도면 되겠지" 했는데, 결국 스케일 아웃을 거치면서 1,000대까지 늘어났다.

전체 구조

[IaC]          Terraform → VPC · EKS · KDS · IAM (코드 한 줄로 전체 인프라 재현)
                                    ↓
[STREAMING]    Generator(K8s) → KDS(4 Shard) → Flink(이상탐지) → Alert KDS → Lambda → Slack
                                    ↓ Firehose
[LAKEHOUSE]    S3 Bronze → S3 Silver → S3 Gold   (Medallion, Parquet + Glue Catalog)
                                    ↓ Airflow ETL
[BATCH]        Bronze→Silver(정제) → Silver→Gold(집계) + 주간 SageMaker ML 재학습
                                    ↓
[ANALYTICS]    Athena(Partition Projection) · SageMaker 예측 · Bedrock AI 채팅 · Grafana

레이어를 5개로 나눴다. 데이터 엔지니어링에서 자주 다루는 5가지 영역을 이 프로젝트에 하나씩 매핑한 구조다.

1. IaC — Terraform으로 전체 인프라 코드화. terraform apply 한 번으로 VPC부터 EKS, KDS, IAM까지 재현 가능하게
2. Streaming — Kinesis & Flink로 실시간 수집 + 윈도우 기반 이상 탐지. Watermark로 Late Data까지 처리
3. Lakehouse — Medallion Architecture(Bronze/Silver/Gold)로 원본 보존과 정제·집계를 레이어별로 분리
4. Batch — Airflow가 멱등성 보장하며 일 단위 ETL 스케줄링 + 주간 SageMaker 모델 재학습
5. Analytics — Athena Partition Projection으로 Ad-hoc 쿼리 비용 70% 절감, SageMaker 예측, Bedrock AI 채팅, Grafana 대시보드

기술 선택에서 고민했던 것들

Kinesis vs Kafka

처음엔 Kafka를 쓸까 했다. 더 많이 쓰이고 레퍼런스도 풍부하다. 근데 이 프로젝트는 AWS 위에서 돌아가고 관리 오버헤드를 최소화하는 게 목표였다. MSK(Managed Kafka)를 쓰면 되긴 하지만 설정이 생각보다 복잡하고, 비용도 예상보다 높았다.

Kinesis Data Streams를 선택한 건 단순한 이유였다. AWS 네이티브라 IAM IRSA 연동이 깔끔하고, Firehose와의 연결이 자연스러웠다. Shard 개수만 잘 계산하면 처리량은 충분했다.

Shard 계산은 이렇게 했다.

로봇 1,000대 × 1 rec/sec = 1,000 rec/sec
레코드 크기: ~200 bytes

KDS 제한:
- 처리량: 1,000 rec/sec per shard
- 대역폭: 1 MB/sec per shard

→ 처리량 기준: 1,000 ÷ 1,000 = 1 shard
→ 안정 마진 고려: 4 Shard로 시작

4 Shard로 시작했다. 나중에 비용 최적화 단계에서 2 Shard까지 줄이는 작업을 하게 되는데, 그 과정에서 꽤 재미있는 사고가 터진다. (4편에서 다룬다)

Flink vs Spark Streaming

실시간 이상 탐지에 Flink를 선택한 이유는 Event Time 처리 때문이었다.

Spark Structured Streaming도 Event Time을 지원하지만, Flink가 Watermark 개념이 더 명확하게 구현되어 있다. 네트워크 지연으로 늦게 도착한 데이터(Late Data)를 어떻게 처리할지 — 이걸 Watermark로 명시적으로 선언할 수 있다는 게 마음에 들었다.

WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND

"10초 이상 늦게 도착한 데이터는 버린다"고 명시하면, Window가 언제 닫히는지 결정론적으로 계산된다. 이게 없으면 Window가 영원히 열려있어서 상태(State)가 계속 쌓인다.

Athena vs EMR vs Spark

일일 배치 ETL에 Athena를 쓴 건 솔직히 가장 쉬운 선택을 한 것이었다.

EMR이나 Spark를 쓰면 더 빠르고 유연하다. 근데 일일 배치는 속도가 크게 중요하지 않았다. Serverless라서 서버 관리가 없고, S3 Parquet과 Glue Data Catalog를 그대로 쓸 수 있었다.

문제는 비용이었다. 처음에 Athena 비용을 계산해보니

일일 데이터량: ~21.6 GB (Parquet 압축 후)
Bronze 스캔: 21.6 GB × $6.25/TB = $0.135
Silver 스캔: 21.6 GB × $6.25/TB = $0.135
→ 월 $280 이상

이걸 줄이는 핵심이 Partition Projection이었다. 파티션 메타데이터를 S3에서 읽지 않고 DDL에 수식으로 정의해두면, Athena가 필요한 파티션만 정확히 스캔한다. 실제로 적용하고 나서 스캔량이 약 70% 줄었다.

인프라 기반 — Terraform + EKS + Karpenter

전체 인프라는 Terraform으로 코드화했다. terraform apply 한 번으로 VPC부터 EKS, Kinesis, IAM까지 생성된다.

EKS 노드 오토스케일링은 Karpenter를 썼다. 기존 Cluster Autoscaler는 Pod 수를 조정할 뿐, 노드 타입까지 최적화하지 않는다. Karpenter는 Pod의 리소스 요구사항에 맞는 가장 저렴한 인스턴스를 자동으로 선택한다. Spot 인스턴스까지 활용하면 On-Demand 대비 70% 절감이 가능하다.

IRSA(IAM Roles for Service Accounts)도 처음 도입해봤다. Pod에 AWS 자격증명을 .env로 박는 게 아니라, Kubernetes ServiceAccount에 IAM Role을 연결해서 임시 자격증명을 자동 주입하는 방식이다. Generator Pod이 KDS에 데이터를 보낼 때, 별도의 AWS_ACCESS_KEY 없이 IRSA 권한만으로 동작한다.

CI/CD는 GitHub Actions + OIDC로 구성했다. Actions에서 AWS 자격증명을 시크릿으로 관리하는 대신, OpenID Connect 토큰을 발급받아 AWS STS에서 임시 자격증명을 받는다. access key를 어딘가에 저장할 필요가 없어진다.

스케일 아웃의 흐름

이 프로젝트는 처음부터 1,000대를 목표로 한 게 아니었다.

초기:  로봇 100대,  KDS 1 Shard
↓
중간:  로봇 500대,  KDS 2 Shard
↓
최종:  로봇 1,000대, KDS 4 Shard

각 단계마다 병목이 다른 곳에서 나타났다. 처음엔 Generator의 asyncio 코루틴 수가 문제였고, 그 다음엔 KDS Shard throttle이 문제였고, 마지막엔 Firehose buffer 설정이 문제였다.

이 과정에서 생긴 사고들이 나중에 해커톤 PRISM의 안정성 설계에 직접적인 영감이 됐다. "실제로 터져봐야 안다"는 걸 몸으로 배웠다.

2편에서는 데이터 수집부터 실시간 이상 탐지까지 — 센서 데이터 1개가 Slack Alert이 되는 6초의 여정을 따라간다.

데이콘 스마트 제조 AI 해커톤 본선 (2025.05.22) 참가 후기 인과추론 + Multi-Agent로 제조 현장 문제를 풀어본 이야기

시작 전에

해커톤이 끝났다.

사실 제조 공정 쪽은 완전 문외한이었다. 기계가 어떻게 돌아가는지도 몰랐고, OEE가 뭔지도 몰랐고, CNC 머신이 뭔지도 해커톤 준비 전까지는 솔직히 관심도 없었다. 그런데 몇 주를 이 도메인에 파묻히고 나니 신기하게도 제조 공정에서 AI가 어떤 역할을 해야 하는지에 대해 내 나름의 시각이 생겼다.

이 글은 화려한 수상 후기가 아니다. 오히려 개발 과정에서 *"이게 왜 이렇게 복잡해졌지?"* 를 반복하면서 어떤 판단을 했고, 무엇이 진짜 어려웠는지를 기록하고 싶었다.

특히 인프라 파이프라인보다 AI 두뇌 부분 — Multi-Agent 협상 구조, 인과추론, Closed-Loop 학습 — 에 집중해서 정리해보려 한다.

뭘 만들었나

PRISM (Process Reasoning & Intelligent Supervision for Manufacturing)

핵심 문제 정의는 이거였다.

메이커스페이스나 중소 제조업체에서 1인 운영자가 공정 이상이 생기면 원인 분석에 1~2시간을 쓴다. 대형 MES 시스템을 쓰면 해결되지만 연간 1,000만 원 이상이다. 우리는 노트북 한 대, 월 2만 원으로 그 격차를 좁히는 걸 목표로 했다.

그 수단이 Closed-Loop AI 였다.

센서 통합 → 인과 RCA → Multi-Agent 협상 → 학습 자산화
         ↑___________________________________|

"Closed-Loop"이라는 단어를 쓴 건, 단순히 이상 탐지하고 끝나는 게 아니라, 발생한 이상이 다시 모델 개선으로 이어지는 순환을 만들었기 때문이다.

두뇌 설계: 가장 많이 고민한 부분

인프라(DuckDB, Streamlit, Docker)는 비교적 선택이 명확했다. 진짜 고민은 AI가 어떻게 "생각"하게 만들지 였다.

1. 인과추론을 선택한 이유

처음에는 XGBoost 예측 결과를 보여주면 되지 않을까 생각했다. 근데 곰곰이 생각해보니 문제가 있었다.

XGBoost가 *"공구 마모(TWF) 가능성 62%"* 라고 말해줘도 운영자는 "그래서 뭘 어쩌라고?" 에서 멈춘다.

상관관계와 인과관계의 차이다. spindle_rpm이 높을 때 불량이 많이 나온다고 해서 spindle_rpm을 낮추면 불량이 줄어드는 게 아닐 수 있다. 운영자가 실제로 필요한 건 "내가 어떤 조치를 취하면 불량이 줄어드느냐" 다.

그래서 DoWhy 라이브러리로 6-Node 인과 DAG를 구축했다.

tool_age ──→ vibration_xyz ──→ dimension_dev ──→ DEFECT
tool_age ──→ thermal_drift ──→ dimension_dev
spindle_rpm → vibration_xyz
spindle_rpm → coolant_temp

여기서 do-calculus 개입(intervention) 을 쓴다. do(tool_age=0) — 즉 *"공구를 지금 교체했다고 가정하면"* — 불량률이 얼마나 달라지는지 ATE(Average Treatment Effect)로 추정해서 운영자에게 보여준다.

모델 신뢰도도 Wright(1991) partial R²를 기반으로 σ_max 임계값을 설정했다.

이걸 UI에 바로 노출하지 않고 "🔍 자세히" expander 를 달아서, 심사위원이 물어볼 경우에 대비해 DoWhy 반박 추정(refutation) 원문을 열람할 수 있게 했다.

*"우리 모델이 완벽하다"고 포장하지 않고 "이 정도 불확실성이 있다"는 걸 투명하게 보여주는 게 더 정직한 접근이라고 생각했다.*

2. Multi-Agent 협상 구조 — 왜 에이전트 4개인가

이상이 탐지됐을 때 어떤 조치를 취해야 하는지는 한 관점만으로는 결정하기 어렵다.

이 네 관점은 자주 충돌한다. 생산 에이전트는 "계속 가자"고 하고, 안전 에이전트는 "당장 멈춰야 한다"고 한다.

Claude Sonnet을 Supervisor로, Haiku를 Domain Agent 4개로 쓴 이유가 여기 있다. Haiku가 각 도메인에서 빠르게 분석을 내놓으면, Supervisor가 이걸 종합해서 Single-Scalar Net Value (KRW) 로 결정을 내린다.

net_value_KRW = throughput_gain
              - α × defect_loss
              - β × safety_loss    # β default=1.0, 기준값 1억 KRW
              - γ × rul_loss

안전 위반이 1억 원 페널티인 건 의도적이다. Streamlit 슬라이더로 β를 1.0 → 2.0으로 올리면 Supervisor 결정이 바뀌는 걸 실시간으로 보여줬다.

"AI가 어떤 가중치를 쓰는지 운영자가 직접 조정할 수 있어야 한다"

Hard-block 규칙도 있다. safety.estop_required = True 이면 해당 선택지의 net_value는 자동으로 -∞ 가 된다. 어떤 상황에서도 비상정지가 필요한 경우에는 경제적 계산 이전에 차단된다.

3. Closed-Loop 학습 — 가장 까다로웠던 부분

불량 #47이 실제 발생한 후, 그 패턴을 XGBoost 모델에 추가 학습시켜 accuracy 0.81 → 0.97 로 올라가는 걸 라이브로 보여주는 게 목표였다.

문제는 "라이브"와 "결정론적"이 충돌한다는 거였다.

시연 중에 랜덤 요소가 끼어들면 어떤 날은 0.97이 나오고 어떤 날은 0.94가 나온다. 해커톤 당일 평가자 앞에서 수치가 달라지면 안 됐다.

그래서 Triple Insurance 를 만들었다.

① 시드 고정
   PYTHONHASHSEED=2026
   random.Random(2026)
   np.random.seed(2026)

② LLM 캐시 재생
   SHA256 해시 키 → JSONL 저장
   PRISM_MODE=demo → Bedrock 호출 0회

③ 영상 자동 fallback
   네트워크 장애 / 타임아웃 →
   사전 녹화 mp4 자동 전환

PRISM_MODE=demo 환경변수 하나로 이 모드가 활성화된다. D-1 Verification Gate 결과: cache_hit_rate = 1.000, bedrock_tokens = 0. 완벽히 오프라인 재현 가능한 상태였다.

개발하면서 제일 골치 아팠던 것들

솔직하게 적는다.

DoWhy + networkx 버전 충돌

가장 예상 못한 복병이었다. DoWhy 0.8이 nx.algorithms.d_separated를 참조하는데, networkx 3.4+에서 이게 nx.algorithms.d_separation.is_d_separator로 이름이 바뀌었다. 에러 메시지도 직관적이지 않아서 처음엔 뭐가 문제인지 한참 헤맸다.

결국 사이트 패키지를 건드리지 않고 프로젝트 레벨에서 monkey-patch로 해결했다.

if not hasattr(nx.algorithms, "d_separated"):
    nx.algorithms.d_separated = nx.algorithms.d_separation.is_d_separator

오픈소스 버전 호환성 문제는 항상 이런 식으로 찾아온다는 걸 다시 배웠다.

Narrative Drift

코드가 바뀌면 발표 대본, 운영자 가이드, README, 시연 스크립트 — 이 네 곳을 동시에 갱신해야 한다.

개발 막바지에 UI에는 "OEE +32%p"라고 나와 있고 발표 슬라이드에는 "+35%"가 남아있는 걸 발견했다. 수치가 다른 걸 심사위원이 잡아내면 신뢰도가 한 번에 무너진다.

마지막 이틀은 코드보다 문서 정합 작업에 더 많은 시간을 쓴 것 같다.

Streamlit 실행 흐름

버튼 클릭마다 전체 스크립트가 재실행되는 구조라, 마커 전환 로직을 어디에 두느냐에 따라 "다음 마커" 버튼이 클릭 이벤트를 먹어버리는 문제가 생겼다. CNC 스트림 루프를 main() 끝으로 옮기고서야 해결됐다.

이번 해커톤으로 늘어난 도메인 지식

제조 공정을 거의 모르고 시작했는데, 어느 순간 이런 개념들이 자연스러워졌다.

OEE (Overall Equipment Effectiveness) 세계 평균 약 0.60인데, 우리 시연은 0.34 → 0.67 시나리오를 구성했다. Nakajima 기준 절대값이라서 단순한 "32%p 향상"이 아니라 맥락이 있는 수치다.

공구 마모 유형 TWF(Tool Wear), HDF(Heat Diffusion), PWF(Power Failure), OSF(Overstrain)... XGBoost를 6-class 분류기로 학습시키면서 각 고장 유형이 어떤 센서 패턴과 연결되는지를 이해하게 됐다.

RUL (Remaining Useful Life) 설비 잔여 수명. 공구를 지금 교체할 것인지 나중에 교체할 것인지의 경제적 트레이드오프 계산에서 핵심 변수가 된다. rul_hour_cost = ₩25,000/h 를 기준값으로 쓴 게 이 맥락이다.

제조 도메인을 모른 채 "AI를 적용하면 좋겠다"로만 접근했다면, 이 수치들이 의미하는 바를 제대로 설계할 수 없었을 것이다.

마치며

첫 해커톤이라 잘 모르는 게 너무 많았다. 기획서를 처음 봤을 때 어디서부터 시작해야 할지도 막막했고, 인과추론이나 Multi-Agent 같은 개념을 실제 제품에 녹여내는 경험도 처음이었다.

그럼에도 이번에 제일 잘한 결정은 하나였다.

"파이프라인보다 두뇌를 먼저 설계하자"

데이터를 어떻게 흘릴지보다, AI가 어떤 논리로 판단하는지를 먼저 정의하니까 나머지가 그 구조에 맞게 자리를 찾았다. Closed-Loop라는 개념이 그 중심을 잡아줬다.

결과가 어떻든, 공장이라는 물리 세계에서 인과추론이 어떤 역할을 할 수 있는지를 스스로 설계하고 구현해본 경험은 오래 남을 것 같다.

마지막으로 해커톤에서 찍은 사진 몇개 첨부하고 긴글 읽어줘서 감사하고 계속 블로그 업데이트해보도록 노력해보겠다.

본 포스팅에서는 로컬 환경에서 로봇 센서 데이터를 수집 및 가공하고, Apache Kafka를 거쳐 최적화된 포워더(Vector)를 통해 AWS S3로 적재하는 엔드투엔드(End-to-End) 데이터 파이프라인의 구축 과정과 기술적 의사결정을 정리한다.

1. 설계 목표 및 파이프라인 아키텍처 본 아키텍처의 핵심은 데이터 형식에 따른 처리 채널의 분리와 경량화된 클라우드 전송(Egress)이다.

(캡처 권장: 파이썬 소스부터 Fluent Bit, Kafka, Vector를 거쳐 S3까지 이어지는 전체 구조도)

파이프라인은 다음과 같은 논리적 흐름을 갖는다.

Source: log_gen.py를 통한 두 가지 형태(JSON, Text)의 로그 파일 생성.

Ingestion: Fluent Bit를 활용한 파일 Tailing 및 라우팅.

Transform: Logstash를 거쳐 비정형 텍스트를 구조화(Grok 파싱).

Messaging: KRaft 모드 기반의 다중 토픽 Kafka 클러스터.

Egress & Storage: Vector를 통한 Kinesis Data Firehose 데이터 포워딩 및 최종 Amazon S3(JSON 형태) 적재.

2. Apache Kafka의 본질과 KRaft의 도입 Kafka는 단순한 메시지 큐가 아니라 데이터를 지연 없이 흘려보내는 이벤트 스트리밍 플랫폼이다. 데이터를 생산하는 프로듀서(Producer)와 이를 소비하는 컨슈머(Consumer) 간의 의존성을 완벽히 분리한다.

(캡처 권장: factory-json-topic과 factory-text-topic, 그리고 내부 상태 관리용 토픽들이 보이는 화면)

이번 아키텍처에서는 기존 Zookeeper 기반의 구성을 탈피하고 KRaft(Kafka Raft) 모드를 도입했다. 과거 수백만 개의 파티션 환경에서 컨트롤러 병목을 유발하던 Zookeeper 의존성을 제거함으로써, 메타데이터를 내부 로컬 로그로 관리하여 아키텍처를 단순화하고 로컬 컨테이너 리소스 점유율을 최적화했다.

3. 수집 및 가공: Channel 분리와 Logstash Filter의 역할 모든 데이터를 동일한 파이프라인으로 처리하는 것은 비효율적이다. 본 설계에서는 원본 데이터의 구조화 여부에 따라 라우팅을 이원화했다.

[ LAYER ]          [ COMPONENT ]          [ PROTOCOL / ACTION ]            [ DATA FORMAT ]
====================================================================================================
SOURCE       :     Python (log_gen.py)  ----( Write to File )---->  [ /sensor_logs/*.log ]
               |                                                                   |
               |                                                            ( Shared Volume )
               v                                                                   |
INGESTION    :     Fluent Bit (Agent)   <---( Tailing File  )----------------------+
               |          |
               |          | [ CHANNEL A: DIRECT BYPASS(json) ]
               |          +--------------------------------------------( Produce )--+
               |          |                                                         |
               |          | [ CHANNEL B: PROXY ROUTE(text) ]                        |
               |          +----( Forward )----> [ Logstash ] ----( Produce )--+     |
               |                                     |                        |     |
               v                                     v                        v     v
TRANSFORM    : (A: Bypass)  +--[ Grok    : Pattern Matching & Extract Fields   ]--+ |
               |            +--[ Mutate  : Data Type Conversion (Str->Float)   ]--+ |
               |            +--[ Tagging : Add Metadata for Conditional Alerts ]--+ |
               |                                                              |     |
MESSAGING    :     Kafka / MSK (Broker) <-------------------------------------+-----+
               |          |             ( Merge Structured & Processed Data )
               |          |
               |          +----( Topic A: factory-json-topic ) <--- [Channel A Result]
               |          +----( Topic B: factory-text-topic ) <--- [Channel B Result]
               v                                                               |
               v                                                               |
                                                                        kafka connect / ui
                                                                                |
                                                                                v               
VISUALIZE    :     OpenSearch (DB)      <---( Indexing & Search )---->  DASHBOARD (Kibana)
=================================================================================================

Channel A (직결): 이미 구조화된 JSON 데이터는 가공 레이어를 생략하고 즉시 factory-json-topic으로 전송하여 Latency를 최소화한다.

Channel B (가공): 비정형 Text 데이터는 Fluent Bit에서 HTTP(포트 5044)를 통해 Logstash로 전달된다.

이 과정에서 HTTP 통신 프로토콜 특성상 막대한 양의 네트워크 메타데이터(IP, 헤더, 버전 등)가 노이즈처럼 결합된다. 예시)

{
    "event":{
        "original":"[{\"date\":1777005819.42045,\"log\":\"\\\"[2026-04-24 13:43:38] ID=AI-FACTORY-001 |   TEMP:87.5 |   HUMI:42.1 |   STAT:RUNNING\\\"\"}]"},
        "host":{"ip":"172.19.0.3"},
        "user_agent":{"original":"Fluent-Bit"},
        "date":1.77700581942045E9,
        "url":{
            "path":"/",
            "domain":"logstash",
            "port":5044
        },
        "http":{
            "version":"HTTP/1.1",
            "method":"POST",
            "request":{
                "body":{
                    "bytes":"124"
                },
                "mime_type":"application/json"
            }
        },
        "log":"\"[2026-04-24 13:43:38] ID=AI-FACTORY-001 |   TEMP:87.5 |   HUMI:42.1 |   STAT:RUNNING\"","@version":"1",
        "@timestamp":"2026-04-24T04:43:40.412003029Z"
    }

이러한 오염된 데이터를 그대로 Kafka에 적재하면 하위 스토리지 비용 증가 및 쿼리 복잡도를 유발한다. 따라서 Logstash의 filter 블록 내에서 grok 패턴([%{TIMESTAMP_ISO8601}] ID=%{DATA}...)을 엄격하게 적용하여, 필요한 센서 지표만 추출한 정제된 구조화 데이터로 변환 후 factory-text-topic으로 발행했다.

4. 클라우드 전송 아키텍처: 왜 Kafka Connect 대신 Vector인가? 로컬 Kafka에 모인 데이터를 AWS 클라우드(Firehose -> S3)로 보내기 위해 초기에는 Kafka Connect를 고려했으나, 엔지니어링 관점에서 명확한 한계가 존재했다.

Java 의존성이 높은 Kafka Connect는 무겁고, 플러그인 관리 및 API 기반 설정이 번거롭다. 이를 대체하기 위해 Datadog에서 개발한 Vector를 도입했다.

압도적인 경량화: Rust 기반으로 작성되어 메모리 사용량이 극히 적다. 로컬 개발 환경에서 매우 유리하다.

운영 복잡도 감소: 복잡한 커넥터 설치 없이, 단일 vector.yaml 파일 하나로 파이프라인(Kafka -> Vector -> Firehose) 구성이 완료된다.

5. 데이터 적재 확인 및 파이프라인 검증 Vector를 통해 포워딩된 데이터는 AWS Kinesis Data Firehose의 버퍼링(크기/시간 조건)을 거쳐 최종적으로 S3 버킷에 JSON 형태로 안전하게 적재된다.

로컬 디렉토리에서 시작된 단 한 줄의 텍스트 로그가 경량 에이전트, 스트리밍 허브, 포워더를 거쳐 클라우드 스토리지에 무손실 적재되는 전체 라이프사이클을 검증했다.

6. 마치며 로컬과 클라우드를 잇는 하이브리드 스트리밍 아키텍처가 안정적으로 구축되었다. 불필요한 시스템(Zookeeper, Kafka Connect)을 도려내고, 각 계층에 가장 효율적인 도구(KRaft, Vector)를 배치하여 파이프라인의 체급을 낮추면서도 처리량은 극대화했다.

S3에 적재된 JSON 데이터는 향후 Amazon Athena를 활용한 서버리스 SQL 쿼리 분석의 기반이 될 것이며, 실시간 탐지가 필요한 구간에는 Flink를 추가 배치하여 파이프라인을 확장해 나갈 예정이다.

데이터 레이크하우스 아키텍처인 메달리온 아키텍처(Medallion Architecture)를 구현하며, S3에 저장된 Raw 데이터를 Bronze 계층(Athena/Glue)으로 구조화하는 작업을 진행했습니다.

• 저장 포맷: Parquet (Snappy Compressed)
• S3 경로 구조: s3://.../bronze/year=2026/month=04/day=21/hour=15/
• 작업 내용: Glue Data Catalog에 테이블 스키마를 미리 생성한 후, Athena에서 MSCK REPAIR TABLE 명령어를 통해 S3 데이터를 로드하려 함.

2. 문제 발생

S3에는 분명히 파티셔닝된 데이터가 존재하고, 테이블의 LOCATION도 올바르게 설정했음에도 불구하고 Athena에서 조회 시 결과가 빈 값(Empty Result)으로 나오는 현상이 발생했습니다.

발견된 증상:

1. MSCK REPAIR TABLE 명령어를 실행해도 새로운 파티션이 추가되었다는 메시지가 뜨지 않음.
2. Glue 테이블 설정의 'Partitions' 탭이 비어 있음.
3. 스키마에는 파티션 키(year, month, day, hour)가 설정되어 있으나, 실제 데이터와 연결되지 않음.

3. 원인 분석

문제의 핵심은 "파티션 키의 정의 순서와 S3 디렉토리 계층 구조의 불일치"에 있었습니다.

Athena의 파티션 인식 원리

Athena의 MSCK REPAIR TABLE 명령어는 S3의 폴더 구조를 정해진 순서대로(Hierarchy) 타고 내려가며 탐색합니다.

• 실제 S3 구조: year -> month -> day -> hour
• 문제의 기존 테이블 설정: hour (Partition 0) -> year (Partition 1) -> ...
• 충돌 발생: Athena는 S3의 최상위 폴더가 hour=XX일 것이라고 예상했지만, 실제로는 year=2026 폴더가 먼저 나타나자 탐색을 중단하고 파티션을 인식하지 못한 것입니다.

4. 해결 방법

파티션 키의 순서를 S3의 물리적 경로 순서와 일치하도록 수정했습니다.

[Step 1] 테이블 스키마 수정

Glue Catalog에서 파티션 번호(Index)를 조정하여 S3 계층 구조와 일치시킵니다.

• 수정 전: hour(0), year(1), month(2), day(3)
• 수정 후: year(0), month(1), day(2), hour(3)

[Step 2] Athena 메타데이터 복구

순서가 교정된 상태에서 파티션을 다시 로드합니다.

-- 1. 테이블의 위치 재설정 (확인용)
ALTER TABLE de_06_ai_ma_bronze_db.bronze_tbl 
SET LOCATION 's3://de-ai-30-827913617635-ap-northeast-2-an/medallion/bronze/';

-- 2. 파티션 구조 재스캔 및 등록
MSCK REPAIR TABLE de_06_ai_ma_bronze_db.bronze_tbl;

결과 메시지:

Repair: Added partition to metastore bronze_tbl:year=2026/month=04/day=21/hour=15 (성공적으로 메타데이터에 등록됨)

5. 요약 및 교훈

1. 계층 구조의 중요성: Athena/Glue에서 파티션 키를 설정할 때는 반드시 S3의 물리적 폴더 깊이 순서대로 정의해야 한다.
2. MSCK REPAIR의 한계: 단순히 키 이름이 같다고 인식하는 것이 아니라, 순서가 틀리면 "탐색 실패"로 간주한다.
3. 데이터 포맷 명시: STORED AS PARQUET와 같은 포맷 설정이 실제 파일과 일치해야 하며, SET LOCATION은 메타데이터 주소만 바꿀 뿐 기존 파티션의 상세 경로까지 자동으로 갱신해주지 않으므로 주의해야 한다.

마무리

메달리온 아키텍처의 첫 단추인 Bronze 계층 구축에서 흔히 하는 실수였지만, 이번 트러블슈팅을 통해 Glue 메타데이터와 S3 물리 저장소 간의 매핑 원리를 깊이 있게 이해할 수 있었습니다.

### 1. 문제 상황 (Problem)

ASAC 2기 데이터 엔지니어링 실습으로 S3를 Data Lake로 삼고 AWS Athena를 연동하는 기본 파이프라인을 구축하던 중 오류가 발생했습니다.

• 데이터 원본: S3 S3 data/basic 경로에 다중 JSON(dict 형태)이 나열된 a.txt 파일 업로드.
• 테이블 설정: Athena 쿼리 편집기에서 S3 데이터를 기반으로 외부 테이블(dummy_test_tbl)을 생성.
• 발생 에러: 테이블 생성 후 쿼리 실행 시 아래와 같은 치명적 오류(Fatal Error) 발생.

HIVE_BAD_DATA: Error reading field 'event_id' at position X. invalid magic number

### 2. 팩트 기반 원인 분석 (Root Cause)

이 에러는 S3에 적재된 실제 데이터의 포맷과 Athena(Glue Data Catalog)에 정의된 테이블 포맷 설정이 불일치할 때 발생합니다.

• 잘못된 가정: 테이블을 생성할 때, '분석 효율을 위해 Parquet 포맷을 선택하면 Athena가 알아서 데이터를 Parquet 포맷으로 읽거나 변환해 줄 것'이라는 착각.
• 기술적 팩트 (Magic Number란?): 실제 S3에 있는 파일은 평문 텍스트 형식의 JSON입니다. 그러나 테이블 속성을 PARQUET로 지정하면, Athena의 쿼리 엔진은 파일을 읽을 때 파일의 맨 앞 4바이트에서 Parquet 포맷의 고유 식별자인 매직 넘버(PAR1)를 찾습니다. 텍스트 파일인 JSON에는 이 바이너리 시그니처가 존재하지 않으므로 엔진이 데이터를 해석하지 못하고 즉시 invalid magic number 파싱 에러를 뱉어내는 것입니다. Athena는 쿼리 엔진일 뿐, 테이블 포맷을 바꾼다고 S3 S3 원본 데이터를 변환해주지 않습니다.

### 3. 논리적 해결 방법 (Solution)

데이터의 수집(Raw)과 가공(Refined) 단계를 철저히 분리하여 접근해야 합니다.

Step 1: Raw 데이터 테이블은 원본 포맷과 100% 일치하게 재생성

• 테이블의 파일 포맷을 Parquet가 아닌 원본 데이터와 동일한 JSON으로 정확히 지정하여 테이블을 다시 생성합니다. (내부적으로 OpenX JSON SerDe가 사용됩니다.)
• 이 테이블(dummy_test_tbl_json)을 쿼리하면 정상적으로 S3의 JSON 텍스트를 읽어옵니다.

Step 2: 분석을 위한 Parquet 테이블은 CTAS로 별도 생성 (ETL 처리)

• JSON 포맷은 스캔 비용이 비싸고 성능이 떨어지므로, Parquet 포맷의 데이터 마트가 필요하다면 CTAS (Create Table As Select) 구문을 이용해 물리적 포맷 변환 작업을 수행해야 합니다. ```sql
• • JSON 테이블의 데이터를 읽어 Parquet 포맷으로 S3 새 경로에 저장하는 ETL 로직 CREATE TABLE dummy_test_tbl_parquet WITH ( format = 'PARQUET', external_location = 's3://your-bucket/data/refined/' ) AS SELECT * FROM dummy_test_tbl_json;

    ![](https://velog.velcdn.com/images/mason_dev/post/23349b38-6b10-4107-aa00-f774c06cea84/image.png)
    
    

### 4. 마치며

• 스키마 온 리드(Schema-on-Read): Athena는 데이터를 읽어들이는 시점에 스키마와 포맷을 적용합니다. 물리적인 Storage Format과 논리적인 DDL 포맷 정의가 어긋나면 즉각적인 HIVE_BAD_DATA 장애로 이어집니다.
• 설정 창 클릭 한 번으로 포맷이 바뀌는 것이 아닙니다. 파일의 구조와 엔진의 직렬화/역직렬화(SerDe) 과정에 대한 팩트를 정확히 이해하고 데이터 파이프라인을 설계해야 합니다.

AWS Managed Service for Apache Flink(구 Kinesis Data Analytics) 환경에서 PyFlink를 활용해 Kinesis Data Streams(KDS)의 실시간 데이터를 집계하는 파이프라인을 구축하던 중 발생한 일련의 오류와 해결 과정을 기록한다.

2. 문제 상황 1: Python 버전과 라이브러리 호환성

로컬 환경(Python 3.11)에서 apache-flink==1.15.0 설치 시도 중 subprocess-exited-with-error 발생.

원인 분석

• 팩트: Apache Flink 1.15는 Python 3.10 이상의 환경을 공식 지원하지 않음. 특히 apache-beam 등 의존성 라이브러리가 Python 3.11의 빌드 구조와 충돌함.
• 해결: Python 3.11 환경을 유지하기 위해 Flink 버전을 1.18.0으로 업그레이드하여 설치 진행.

3. 문제 상황 2: 모듈 참조 및 객체 생성 오류

코드 실행 시 ModuleNotFoundError: No module named 'flink' 및 생성자 호출 에러 발생.

원인 분석

• Import 경로: 패키지명은 apache-flink이지만, 실제 코드 내 임포트 경로는 pyflink임. from flink.table은 존재하지 않는 경로.
• 객체 생성: TableEnvironment는 생성자를 직접 호출하지 않고 .create() 정적 메서드를 사용해야 함.

교정된 코드

from pyflink.table import EnvironmentSettings, TableEnvironment

# 잘못된 방식: t_env = TableEnvironment(setting)
# 올바른 방식:
setting = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = TableEnvironment.create(setting)

4. 문제 상황 3: Kinesis Connector 인식 불가 (JAR 버전 mismatch)

CREATE TABLE 구문에서 'connector' = 'kinesis'를 사용했으나, Flink 엔진이 커넥터를 찾지 못함.

원인 분석

• 팩트 1: Flink SQL을 쓰기 위해서는 일반 Connector가 아닌 SQL 전용 통합 JAR(Fat Jar)가 필요함.
• 팩트 2: Flink 1.18 엔진에 1.15용 JAR를 사용하면 클래스 로딩 시 NoSuchMethodError 등 런타임 에러 발생 가능성이 매우 높음.
• 해결: flink-sql-connector-kinesis-4.2.0-1.18.jar 파일로 교체.

5. 문제 상황 4: AWS 환경에서의 배포 및 권한 문제

S3에 코드를 올리고 실행했으나 RestHandlerException 발생 및 데이터 수신 불가.

원인 분석

• EntryPoint 미지정: AWS Managed Flink는 ZIP 내의 어떤 파일이 메인인지 모름. '런타임 속성(Runtime Properties)' 설정 필수.
• IAM 권한 결여: 기본 생성된 역할에는 Kinesis 읽기/쓰기 권한이 빠져 있었음.

최종 설정 데이터

1. Runtime Properties:
   • Group ID: kinesis.analytics.flink.run.options
   • Key: python / Value: app.py
   • Key: jarfile / Value: flink-sql-connector-kinesis-4.2.0-1.18.jar
2. IAM Policy: kinesis:DescribeStream, kinesis:GetRecords, kinesis:PutRecord 등 추가.

6. 결론 및 회고

1. 버전 정합성: PyFlink는 Python, Flink 엔진, Connector JAR의 3박자 버전이 완벽히 맞아야 한다.
2. 환경의 차이: 로컬(Windows)과 배포(AWS Linux) 환경의 경로 체계 차이를 인지하고 하드코딩을 피해야 한다.
3. 권한 확인: "시동이 안 걸리면 설정 문제, 달리다 멈추면 권한 문제"라는 가설을 데이터(CloudWatch)로 증명하는 과정이 중요하다.

Reference

• Apache Flink Documentation (v1.18)
• Maven Repository: flink-sql-connector-kinesis

이 문제를 해결하기 위해 중간에 버퍼(Buffer) 역할을 해주는 Amazon Data Firehose(ADF)를 배치한다. 오늘은 Python으로 가상의 로그를 생성하고, Firehose를 통해 일정한 조건(크기/시간)에 맞춰 S3에 덩어리(Batch) 단위로 적재하는 파이프라인을 구축해 본다.

1. 아키텍처 및 파이프라인 흐름

전체적인 데이터의 흐름은 다음과 같다.

1. Producer: Python Faker 라이브러리를 이용해 금융/이커머스 등 가상 도메인 로그 생성
2. Delivery: Boto3 (AWS SDK)를 통해 Firehose 스트림으로 데이터 전송 (put_record)
3. Buffering & Sink: Firehose가 지정된 버퍼 조건(예: 1MiB or 60초)을 채운 뒤 S3에 파일로 저장

2. 가짜 데이터 생성 및 직렬화 (Log Generator)

데이터 파이프라인 테스트를 위해 가장 먼저 할 일은 현실성 있는 데이터를 만드는 것이다. Faker 라이브러리를 활용해 금융(Finance) 트랜잭션 로그를 생성하는 클래스를 구성한다.

# log_generator.py (일부 발췌)
from faker import Faker
from datetime import datetime
import random

fake = Faker('ko_KR')

class LogGenerator:
    def finance(self):
        return {
            "timestamp" : datetime.now().isoformat(),
            "industry"  : "FINANCE",
            "user_id"   : fake.uuid4()[:12],
            "transaction" : random.choice(['이체','출금','입금','결제']),
            "amount" : round(random.uniform(1000, 1000000), -2),
            "status" : random.choices(["SUCCESS", "FAIL"], weights=[0.95, 0.05])[0]
        }

이 객체 형태의 데이터를 AWS로 전송하려면 문자열 형태(JSON str)로 변환하는 직렬화(Serialization) 과정이 필수적이다.

# run.py
import json

def make_one_log():
    # ensure_ascii=False 옵션으로 한글 깨짐 방지
    return json.dumps(log_gen.finance(), ensure_ascii=False) 

3. Boto3를 이용한 AWS Firehose 연결 (인증)

이제 Python 스크립트가 AWS 인프라에 접근하기 위한 '통행증'을 발급받아야 한다. boto3 라이브러리를 사용해 Firehose 클라이언트를 생성한다.

여기서 가장 중요한 것은 코드가 실행되는 환경에 따른 인증 방식의 분리다. 하드코딩된 자격 증명(Access Key)은 보안 사고의 주범이므로 절대 피해야 한다.

# adf_direct_data_put.py
import boto3

REGION = 'ap-northeast-2'

def get_client(service_name='firehose', is_in_aws=True): 
    # 1. 로컬 환경 (is_in_aws=False)
    # 로컬에 설정된 aws configure 프로필이나 환경변수를 통해 인증
    if not is_in_aws:
        session = boto3.Session(region_name=REGION)
        return session.client(service_name)    

    # 2. AWS 내부 환경 (is_in_aws=True)
    # EC2, CloudShell 등 IAM Role이 부여된 환경에서는 키 입력 없이 자동 인증
    return boto3.client(service_name, region_name=REGION)

# 로컬에서 테스트할 경우 get_client('firehose', False) 로 호출
firehose = get_client() 

4. 데이터 전송 (put_record)

클라이언트 객체가 생성되었으면, 직렬화된 로그 데이터를 Firehose Delivery Stream으로 쏜다.

def send_log():
    response = firehose.put_record(
        DeliveryStreamName='de-ai-06-an2-kdf-log-to-s3',
        Record={
            # 주의: 데이터의 끝을 알리는 개행문자(\n)를 반드시 추가해야 함
            'Data': make_one_log() + "\n" 
        }
    )
    print(f'전송결과 : {response}') # HTTP 200이면 정상

• 엔지니어링 포인트: S3에 적재된 파일을 나중에 읽어 들일 때, 각 로그 객체가 줄바꿈으로 구분되어 있어야 (JSON Lines 형태) 파싱 에러가 나지 않는다. 따라서 'Data': make_one_log() + "\n" 처리는 사소하지만 매우 중요한 부분이다.

5. Firehose의 핵심: 버퍼링 (Size & Interval)

Firehose가 받은 데이터를 즉시 S3에 넣지 않는다고 했다. 그렇다면 언제 넣을까? AWS 콘솔에서 Firehose를 생성할 때 설정한 Buffer size(크기)와 Buffer interval(시간) 조건에 따른다.

• Buffer Size (예: 1MiB): 전송된 데이터가 버퍼 메모리 내에 1 메가바이트만큼 쌓이면 S3로 Flush(전송)한다.
• Buffer Interval (예: 60초): 데이터가 용량을 다 채우지 못했더라도, 첫 데이터가 들어온 지 60초가 경과하면 S3로 Flush한다.

"둘 중 하나라도 먼저 조건을 만족하면(Whichever happens first)" S3로 데이터가 압축되어(또는 원본 그대로) 업로드된다. 이를 통해 파일의 크기를 적절히 키워 S3에 저장함으로써 다운스트림(Downstream)에서의 데이터 처리 효율을 극대화할 수 있다.

1. TypeError: string indices must be integers, not 'str'

문제 상황 신용 평가 API 호출 결과를 파싱하여 DB 적재 함수(_load_users_credit)로 전달하는 과정에서 발생했다. data['user_id']와 같이 딕셔너리 키로 접근하려 했으나 에러가 발생하며 태스크가 실패했다.

원인 분석 에러 메시지는 data 변수가 딕셔너리가 아닌 문자열(String)임을 가리키고 있었다.

• 가정: API 응답이나 이전 태스크에서 넘어온 값이 JSON 객체(dict)일 것이라 판단했다.
• 팩트: 실제 데이터는 역직렬화(Deserialization)되지 않은 Raw String 상태였다. Python에서 문자열에 대괄호([])로 접근하면 인덱스(정수)를 기대하지만, 문자열 키를 넣었기 때문에 발생한 문법 오류였다.

해결 방법 데이터를 사용하는 시점에서 json.loads()를 통해 딕셔너리로 변환하거나, 데이터를 넘겨주는 시점에서 파이프라인의 데이터 직렬화 상태를 점검하여 객체 타입으로 전달되도록 수정했다.

2. MySQLdb.IntegrityError: Duplicate entry for key PRIMARY

문제 상황 데이터 타입 이슈 해결 후, 실제 DB에 데이터를 INSERT 하는 과정에서 발생했다. 특정 유저 아이디(C001)가 이미 테이블의 기본키(Primary Key)로 존재하여 중복 삽입이 불가능하다는 에러였다.

원인 분석

• 설계 결함: 해당 DAG는 주기적으로 신용 점수를 업데이트해야 한다. 하지만 초기 코드는 단순 INSERT 쿼리로 작성되어 있어, 이미 데이터가 존재하는 유저의 경우 제약 조건 위반으로 처리가 중단되었다.
• 테이블 구조: customers 테이블의 PK가 user_id로 설정되어 있어, 동일 유저에 대한 다중 레코드 삽입이 차단된 상태였다.

3. ON DUPLICATE KEY UPDATE를 활용한 해결

단순히 에러를 회피하는 것이 아니라, 비즈니스 로직에 맞게 Upsert(Update + Insert) 방식을 채택하기로 했다.

수정된 SQL 로직

INSERT INTO customers (user_id, credit_score, grade)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE
    credit_score = VALUES(credit_score),
    grade = VALUES(grade);

결과 및 기대 효과

1. 멱등성(Idempotency) 확보: 동일한 DAG를 여러 번 실행해도 에러 없이 최신 데이터로 유지된다.
2. 효율성: 데이터 존재 여부를 먼저 조회(SELECT)하고 조건문으로 분기할 필요 없이, DB 레벨에서 한 번의 쿼리로 삽입과 갱신을 동시에 처리한다.

4. 마치며

이번 이슈를 통해 두 가지 교훈을 얻었다. 첫째, 데이터의 타입을 맹신하지 말고 항상 로그를 통해 객체의 형상을 확인해야 한다. 둘째, DB 적재 로직을 설계할 때는 해당 테이블이 이력(History) 관리용인지, 현재 상태(Master) 관리용인지를 명확히 구분하여 적절한 쿼리 전략을 세워야 한다.

최근 스마트팩토리 온도 센서 데이터를 MySQL로 적재하는 ETL 파이프라인 실습을 진행하며 겪은 치명적인 오류와 그 해결 과정을 공유합니다.

비슷한 형태의 파이프라인 여러 개를 테스트하기 위해 코드를 여러 파이썬 파일(file_A.py, file_B.py)로 나누어 작성했습니다. 이때, "동일한 dag_id를 부여하면 AWS의 로드밸런서(ELB)처럼 스케줄러가 여러 DAG를 라운드로빈 방식으로 분산 처리해 주지 않을까?"라는 가설을 세웠습니다.

하지만 결과는 처참한 파이프라인 붕괴였습니다.

Symptoms (발생한 증상)

대시보드와 실행 로그에서 다음과 같은 기괴한 현상들이 발생했습니다.

1. 대시보드의 깜빡임(Flickering) 현상: 웹 UI에서 새로고침(F5)을 누를 때마다 DAG의 구조나 태스크 목록이 계속 다른 파일의 형태로 바뀌어 보였습니다.
2. Task Failed 및 무한 Retry 지옥: 워커가 태스크를 실행하려다 말고 갑자기 Task not found in DAG라는 에러를 뱉으며 실패했습니다. Airflow는 재시도(Retry)를 지시하지만, 영원히 성공하지 못하고 상태가 꼬여버렸습니다.

Root Cause Analysis (원인 분석)

결론부터 말하자면, 저의 가설은 Airflow의 시스템 아키텍처를 오판한 치명적인 착각이었습니다. Airflow는 DAG 단위로 로드밸런싱을 수행하는 네트워크 도구가 아닙니다.

이 사태의 근본적인 원인은 '메타데이터 DB의 고유키 충돌'과 '스케줄러의 파싱(Parsing) 방식'에 있었습니다.

1. dag_id는 데이터베이스의 Primary Key(고유키)다

Airflow의 모든 설정과 상태는 메타데이터 DB(Postgres, MySQL 등)에 저장됩니다. 이때 dag 테이블에서 dag_id는 무조건 고유해야 하는 Primary Key(PK)입니다. 동일한 식별자를 가진 파일이 여러 개 존재한다는 것은, 분산 처리가 아니라 데이터 충돌(Collision)을 의미합니다.

2. Scheduler의 덮어쓰기(Overwrite) 경합 (Race Condition)

Airflow 스케줄러는 백그라운드에서 dags/ 폴더 내의 파일들을 주기적으로 순회하며 파싱(Parsing)합니다.

• T1 시점: 스케줄러가 file_A.py를 파싱합니다. DB의 dag_id="my_etl" 정보가 file_A의 구조로 덮어써집니다. (UI에는 A가 보임)
• T2 시점: 몇 초 뒤, 스케줄러가 file_B.py를 파싱합니다. DB의 my_etl 정보가 file_B의 구조로 다시 덮어써집니다. (새로고침하면 UI에 B가 보임)

즉, 로드밸런싱이 되는 것이 아니라 스케줄러가 파싱할 때마다 DB 레코드가 엎치락뒤치락하며 덮어쓰기 전쟁을 하고 있었던 것입니다.

3. 고아 태스크(Orphan Task)의 발생

이 상태에서 Worker가 file_A의 task_1을 실행하려는데, 찰나의 순간 스케줄러가 file_B를 파싱해버렸다고 가정해 봅시다. DB 상의 DAG 구조는 file_B로 바뀌었으므로, Worker는 자신이 실행해야 할 task_1을 DB에서 찾지 못합니다. 그 결과 "DAG 안에 해당 Task가 존재하지 않는다"며 비정상 종료(Failed) 처리되는 것입니다.

Solution & Action Item (해결책)

시스템의 원리를 이해했다면 해결책은 간단합니다.

1. 파일별로 독립적인 dag_id 부여 각 파이썬 파일마다 반드시 고유한 dag_id를 사용해야 합니다. (예: 05_mysql_etl_sensor_A, 05_mysql_etl_sensor_B)

2. Dynamic DAG (동적 DAG) 활용 만약 수십 개의 센서에서 들어오는 데이터를 동일한 로직으로 처리해야 해서 하드코딩이 어렵다면, 하나의 파이썬 파일 내에서 for 문을 활용해 동적 DAG를 생성하는 방식을 채택해야 합니다.

# Dynamic DAG 생성 예시
sensors = ['sensor_A', 'sensor_B', 'sensor_C']

for sensor in sensors:
    dag_id = f"05_mysql_etl_{sensor}"

    with DAG(dag_id=dag_id, ...) as dag:
        # 태스크 정의
        # ...

    # 글로벌 네임스페이스에 DAG 등록 (Airflow가 인식하게 함)
    globals()[dag_id] = dag

📝 Conclusion (마치며)

이번 트러블슈팅을 통해 Airflow의 스케줄러가 파일을 파싱하는 루프 방식과, 메타데이터 DB가 전체 시스템의 중심에서 어떻게 작동하는지 명확히 깨달았습니다.

Error: Already running on PID 42 
(or pid file '/opt/airflow/airflow-webserver.pid' is stale)
```![](https://velog.velcdn.com/images/mason_dev/post/258df9f9-be95-4bf3-bddb-4e7d1b6e4729/image.png)


그리고 `localhost:8080` 접속이 되지 않는 상황.

---

## 📌 문제 원인 분석

Airflow는 웹서버 실행 시 다음 경로에 PID 파일을 생성한다.

/opt/airflow/airflow-webserver.pid

이 파일에는 실행 중인 프로세스 ID(PID)가 기록된다.

하지만 아래와 같은 상황이 발생하면:

* 컨테이너 강제 종료
* `kill -9`
* 비정상 종료
* Docker 재시작 중 충돌

👉 **PID 파일이 삭제되지 않고 남게 된다.**

그 결과 Airflow는:

> “이미 실행 중이다”라고 잘못 판단하고 종료한다.

이를 **stale PID file 문제**라고 한다.

---

## 구조 이해 (추천 이미지)

velog에 아래와 같은 구조 다이어그램 하나 넣으면 이해도가 확 올라간다.

### 📷 추천 이미지 1: Airflow PID 동작 구조

[ airflow webserver 실행 ] ↓ [ airflow-webserver.pid 생성 ] ↓ [ 비정상 종료 ] ↓ [ pid 파일은 남아있음 ] ↓ [ 재실행 시 충돌 발생 ]

(다이어그램 툴: draw.io / Excalidraw 추천)

---

# 해결 방법 1 — 컨테이너 내부에서 직접 해결

### 1️⃣ 실행 중인 컨테이너 확인

```bash
docker ps

2️⃣ 컨테이너 접속

docker exec -it <webserver_container_name> bash

3️⃣ PID 파일 삭제

rm -f /opt/airflow/airflow-webserver.pid

확인:

ls /opt/airflow | grep pid

4️⃣ 웹서버 재실행

airflow webserver

또는 (Airflow 2.x)

airflow webserver --port 8080

해결 방법 2 — docker-compose 사용 시 (권장)

Airflow를 docker-compose로 실행 중이라면 통째로 재기동하는 것이 더 깔끔하다.

docker-compose down
docker-compose up -d

그래도 해결되지 않으면:

docker-compose down -v
docker-compose up -d

⚠ -v 옵션은 볼륨 삭제 → DB 초기화됨

추가 확인 (실제로 프로세스가 살아있는지 체크)

혹시 진짜 웹서버 프로세스가 떠 있는 경우도 있으니 확인:

ps -ef | grep airflow

실행 중이라면 kill 후 재시작.

실무 정리

docker logs <webserver_container_name>

근본적 예방 방법

1. 컨테이너 강제 종료 지양
2. docker stop 사용
3. 운영 환경에서는 systemd 또는 supervisor 사용
4. healthcheck 설정 고려

핵심 한 줄 정리

Airflow Already running on PID 오류는 대부분 stale PID 파일 때문이다.

이러한 문제를 해결하기 위해 에어비앤비(Airbnb)에서 개발하여 오픈소스로 공개한 워크플로우 관리 플랫폼이 바로 Apache Airflow입니다. 오늘은 Airflow의 핵심 개념과 내부 아키텍처에 대해 정리해 봅니다.

1. Apache Airflow란?

Airflow는 복잡한 데이터 파이프라인을 파이썬 코드로 작성하고, 스케줄링 및 모니터링할 수 있는 플랫폼입니다. "Configuration as Code" 원칙을 따르기 때문에, 모든 워크플로우를 파이썬 스크립트로 정의할 수 있어 버전 관리 및 유지보수에 매우 유리합니다.

주요 특징

• 동적 파이프라인: 파이썬 코드를 이용해 동적으로 파이프라인을 구성할 수 있습니다.
• 확장성: 다양한 시스템(AWS, GCP, Hadoop 등)과 연동할 수 있는 수많은 플러그인과 오퍼레이터를 지원합니다.
• 우수한 UI: 작업의 상태, 로그, 실행 시간 등을 웹 대시보드에서 직관적으로 파악할 수 있습니다.

2. 핵심 개념

Airflow를 다루기 위해 반드시 알아야 할 3가지 핵심 요소가 있습니다.

DAG (Directed Acyclic Graph)

Airflow에서 워크플로우를 구성하는 가장 기본이 되는 단위입니다. 수학적 개념인 '방향성 비순환 그래프'를 의미하며, 작업들이 실행되는 순서와 의존성을 정의하지만 절대 순환(Loop) 구조를 가지지 않는 것이 특징입니다.

Operator와 Task

• Operator (오퍼레이터): 수행하고자 하는 작업의 '템플릿' 또는 '클래스'입니다. 어떤 작업을 할 것인지(예: 파이썬 함수 실행, Bash 명령어 실행 등)를 정의합니다. (ex. BashOperator, PythonOperator)
• Task (태스크): 오퍼레이터가 DAG 내에서 인스턴스화되어 실제 실행을 위해 구체화된 객체입니다. DAG를 구성하는 하나의 노드(Node)가 됩니다.

3. Airflow 내부 아키텍처

Airflow가 어떻게 수많은 스케줄을 누락 없이 관리하고 실행하는지 이해하려면 내부 아키텍처를 살펴보아야 합니다. Airflow는 단일 프로그램이 아니라, 여러 컴포넌트가 유기적으로 통신하는 분산 시스템에 가깝습니다.

주요 컴포넌트

1. Scheduler (스케줄러): 가장 핵심적인 데몬입니다. 주기적으로 DAG 폴더를 파싱하여 예약된 작업이 있는지 확인하고, 실행 조건이 충족된 Task를 실행 큐(Queue)에 넣습니다.
2. Executor (익스큐터): 큐에 들어온 Task를 어떻게 실행할지 결정하는 메커니즘입니다. 로컬에서 프로세스를 띄울지(LocalExecutor), Celery를 통해 분산 워커로 보낼지(CeleryExecutor), Kubernetes Pod으로 실행할지(KubernetesExecutor)를 결정합니다.
3. Worker (워커): Executor의 명령을 받아 실제 Task 작업을 수행하는 프로세스입니다.
4. Metadata Database: DAG의 정의, 현재 상태, Task의 실행 이력 등 Airflow의 모든 상태 정보를 저장하는 RDBMS(PostgreSQL, MySQL 등)입니다.
5. Web Server: Metadata DB에 저장된 상태 정보를 읽어와 사용자에게 웹 UI로 제공합니다.

4. Task의 생명 주기 (Lifecycle)

스케줄러에 의해 Task가 실행될 때, Task는 여러 상태(State)를 거치게 됩니다. 이 상태 변화를 이해해야 파이프라인 디버깅이 수월해집니다.

• No status (None): 스케줄러가 인지하기 전의 초기 상태
• Scheduled: 스케줄러가 실행 조건이 충족되었다고 판단한 상태
• Queued: Executor에게 실행을 요청하여 큐에 대기 중인 상태
• Running: Worker가 Task를 할당받아 실제로 실행 중인 상태
• Success / Failed: 작업이 정상적으로 완료되었거나 오류로 종료된 상태

5. 강력한 모니터링 UI

작성한 코드(DAG)가 의도대로 동작하는지, 병목 현상이 발생하는 구간은 없는지 파악하는 것은 데이터 엔지니어링의 핵심입니다. Airflow는 이를 위해 다양한 View를 제공합니다.

• Grid View: 과거의 실행 이력(DAG Runs)을 그리드 형태로 배열하여, 특정 날짜에 어떤 Task가 실패했는지 직관적으로 파악할 수 있습니다.
• Gantt Chart: 작업들의 병렬 처리 상태와 소요 시간을 확인하여 파이프라인의 성능을 최적화(Tuning)할 때 유용합니다.

마무리

Airflow는 코드로 인프라와 워크플로우를 관리한다는 점에서 강력한 유연성을 제공합니다. 단, 구조가 복잡하여 러닝 커브가 존재하고 초기 세팅(Metadata DB, 메세지 브로커 등)에 비용이 든다는 점은 고려해야 합니다. 다음 포스팅에선 vscode에서 직접 DAG를 생성해보기 위해 docker-compose.yaml 파일를 이용하여 docker에 여러개의 컨테이너를 생성해 airflow에 필요한 각각의 프로그램을 설치해 실습해보겠습니다.

인프라 설계 시 보안 강화를 위해 DB 서브넷을 외부 인터넷과 완전히 격리(Isolated)하는 아키텍처를 채택하곤 합니다. 그러나 이 과정에서 고가용성(HA)을 위한 데이터 동기화 메커니즘과 네트워크 엔드포인트의 역할을 혼동하여 불필요한 설계를 추가하는 실수가 빈번히 발생합니다.

• 핵심 의문: "Multi-AZ 환경에서 가용 영역 간 DB 데이터 동기화를 위해 별도의 VPC Endpoint나 네트워크 피어링 설정이 필요한가?"
• 결론: 그렇지 않습니다. 데이터 복제는 사용자가 제어하는 라우팅 영역 밖에서 이루어집니다.

1. DB 데이터 동기화(Replication)의 실제 작동 메커니즘

AWS RDS의 Multi-AZ 구성은 서비스 연속성을 위한 핵심 요소입니다. 하지만 이들의 통신 방식은 일반적인 VPC 내부 통신과는 궤를 달리합니다.

1.1. 작동 원리: 동기식 복제 (Synchronous Replication)

• 인프라 추상화: 데이터 복제는 인프라 관리자가 정의한 서브넷 라우팅 테이블이나 VPC Endpoint를 경유하지 않습니다.
• 전용 백엔드 네트워크: AWS가 관리하는 별도의 내부 전용 백엔드 네트워크를 통해 물리적 데이터 센터 간 실시간 동기화가 수행됩니다.
• 관리 주체: 이는 관리형 서비스(Managed Service)의 영역으로, 사용자가 YAML 파일이나 콘솔에서 통로를 설정하거나 개입할 수 없습니다.

1.2. 결론

DB 노드 간의 복제 트래픽은 사용자의 VPC 트래픽 쿼터나 보안 그룹 설정(Egress)에 영향을 받지 않는 완전 격리된 평면(Control/Data Plane)에서 동작합니다.

2. VPC Endpoint의 역할과 완전 격리의 맹점

인터넷 게이트웨이(IGW)와 NAT 게이트웨이가 없는 '완전 격리 서브넷'은 보안상 유리하지만, 서비스 운영 측면에서는 '고립'이라는 부작용을 낳습니다.

2.1. 도입 필요성

DB 서브넷에서 외부 인터넷(0.0.0.0/0) 경로를 삭제하면 외부 위협은 차단되지만, AWS의 다른 필수 관리 서비스와의 접점도 상실됩니다. 이를 해결하기 위해 VPC Endpoint라는 프라이빗 전용 통로가 필요합니다.

2.2. DB 서브넷 필수 Endpoint 연동 대상

완전 격리 환경에서도 운영을 위해 반드시 고려해야 할 엔드포인트는 다음과 같습니다.

3. 실무 아키텍처 설계 요약 (Best Practice)

안전하고 견고한 격리형 DB 인프라를 설계하기 위한 3대 원칙입니다.

① 고가용성 (High Availability)

2개 이상의 가용 영역(AZ)에 서브넷을 분산 배치하고 RDS Multi-AZ 옵션을 활성화합니다. 이때 데이터 동기화는 RDS 자체 네트워크 인프라에 위임합니다.

② 보안성 (Isolation)

DB 서브넷의 라우팅 테이블에서 IGW 및 NAT Gateway로 향하는 모든 라우팅을 제거하여 외부 노출을 원천 차단합니다.

③ 운영 가능성 (Feasibility)

격리된 환경 내에서 AWS 서비스와 통신할 수 있도록 VPC Endpoint(S3, SSM 등)를 구성하여 백업 및 관리 효율성을 확보합니다.

Editor's Note: 결국 DB 보안의 핵심은 "무조건적인 차단"이 아니라, "필요한 통로(Endpoint)만 선별적으로 열어주는 전략적 격리"에 있습니다. 복제 메커니즘에 대한 오해를 바로잡는 것이 효율적인 아키텍처 설계의 시작입니다.

4. 마무리

만약 자신이 회사에서 aws를 이용해 인프라를 구축해야 될 상황이 왔을 때 단순히 db subnet은 외부와 완전히 차단되야 된다고 생각해 Nat 게이트웨이뿐만아니라 vpc endpoint까지 연결하지 않는다면 그 subnet은 db로써 역할할 수 없는 "외딴섬"이 되버립니다. 지속적인 db 백업이나 보안 업데이트를 위해 AWS 서비스들과 통신하기 위해서 꼭 vpc endpoint가 있어야 된다는 것을 알게 되었습니다.

1. Bastion Host: 보안의 시작이자 끝

• Bastion Host의 정의: 외부 인터넷에서 프라이빗 서브넷에 있는 자원에 안전하게 접근하기 위한 '프록시(Proxy)' 역할의 서버.
• 필요성: 프라이빗 서브넷은 공인 IP가 없어 직접 접속이 불가능하다. 하지만 관리자는 패치나 설정을 위해 접속해야 하므로, 퍼블릭에 '대문(Bastion)'을 하나 두고 이를 통해서만 들어가게 설계한다.

2. 아키텍처 설계: Multi-AZ 3-Tier 구성

오늘 구축한 네트워크의 핵심 설계를 공유합니다.

• VPC CIDR 설계: 10.0.0.0/20을 사용하여 총 4,096개의 IP 확보.
• 서브넷 전략: * Public: 로드밸런서(ALB), NAT 게이트웨이, Bastion 배치.
  • App Private: 애플리케이션 서버 배치. EKS 파드 확장을 고려해 /22(1024개 IP)로 넉넉하게 할당.
  • DB Private: 외부와 완전히 차단된 Isolated 영역.
• 고가용성(HA): 가용 영역(AZ) 1, 2에 각각 NAT 게이트웨이를 배치하여 데이터센터 장애에 대비.

3. 오늘의 삽질: 트러블슈팅 리포트 (핵심 섹션)

이슈 1: "서브넷은 퍼블릭인데, 인스턴스는 퍼블릭이 아니다?"

• 상황: Bastion 호스트에 접속하려니 "퍼블릭 서브넷에 있지 않다"는 에러 발생.
• 원인: CloudFormation의 경쟁 상태(Race Condition). 인터넷 게이트웨이(IGW)가 VPC에 완전히 붙기 전에 인스턴스가 먼저 생성되면서 경로를 찾지 못한 것.
• 해결: DependsOn: VPCGatewayAttachment 속성을 통해 리소스 생성 순서를 명시적으로 제어.

이슈 2: error in libcrypto SSH 접속 실패

• 상황: Bastion에서 프라이빗 서버로 접속 시 .pem 키 로드 에러 발생.
• 원인: 키 파일의 권한(chmod 400) 미비 및 복사 과정에서의 포맷 손상.
• 해결: PuttyGen을 통한 올바른 OpenSSH 포맷 변환 및 권한 설정의 중요성 확인.

4. 아키텍처적 고찰: ALB는 왜 퍼블릭에 있는가?

• 로드밸런서(ALB)는 사용자를 맞는 '정문'이므로 퍼블릭에 위치해야 한다.
• 하지만 실제 데이터가 있는 서버는 '안방(Private)'에 숨긴다.
• 결론: 로드밸런서가 퍼블릭에서 트래픽을 받아 프라이빗 타겟 그룹으로 넘겨주는 구조가 보안의 정석임을 실습으로 증명.

EKS 클러스터에서 EBS CSI Driver를 설정하던 중, 크게 두 가지 단계에서 에러가 발생했습니다.

에러 1: eksctl을 이용한 IAM 서비스 계정(IRSA) 생성 실패

eksctl create iamserviceaccount \
  --name ebs-csi-controller-sa \
  --role-name AmazonEKS_EBS_CSI_DriverRole_new \
  ... --approve

• 증상: waiter state transitioned to Failure, Error: failed to create iamserviceaccount(s)
• 원인: --role-name에 지정한 이름(AmazonEKS_EBS_CSI_DriverRole_new)의 IAM 역할이 이미 AWS 계정에 존재하기 때문입니다. CloudFormation은 동일한 이름의 리소스를 생성하려고 할 때 충돌을 일으키며 롤백됩니다.

에러 2: EKS Addon 생성 후 CREATE_FAILED 및 업데이트 불가

• 증상: aws eks describe-addon 결과 상태가 CREATE_FAILED이며, ConfigurationConflict 메시지 출력. 이를 해결하기 위해 update-addon을 시도했으나 ResourceInUseException 발생.
• 원인: 이미 eksctl을 통해 생성된 ebs-csi-controller-sa가 존재하기 때문에, EKS 애드온 관리자가 해당 리소스에 대한 권한(Managed-by)을 가져오지 못해 충돌이 발생한 것입니다.

*여기서 ebs-csi-controller-sa 란

• Amazon EKS 환경에서 EBS(블록 스토리지) 볼륨의 생명주기를 관리하는 ebs-csi-controller 파드에 IAM 권한을 부여하기 위해 사용하는 쿠버네티스 ServiceAccount

• AWS IAM 역할(IRSA)과 연결되어, 컨트롤러가 볼륨 생성, 연결 등 AWS API 작업을 수행할 수 있도록 인증하는 핵심 요소

2. 원인 분석 (Deep Dive)

가설 1: 리소스 네이밍 충돌

eksctl은 CloudFormation 스택을 통해 작업을 수행합니다. --role-name 옵션을 고정하면, 해당 이름의 IAM Role이 이미 존재할 경우 스택 생성이 실패합니다. IAM Role은 전역 리소스이므로 고유한 이름을 가져야 합니다.

가설 2: 상태 관리의 교착 상태

AWS EKS 애드온의 상태 머신(State Machine)은 매우 엄격합니다.

1. ConfigurationConflict: 서비스 계정(SA)의 관리 주체가 eksctl인지 EKS 애드온인지 결정되지 않아 발생합니다.
2. ResourceInUseException: 애드온이 CREATE_FAILED 상태에 빠지면, AWS API는 이를 '불완전한 리소스'로 간주하여 update 명령을 허용하지 않습니다. 오직 delete 후 다시 create 하는 것만 허용됩니다.

3. 해결 방법 (Step-by-Step)

Step 1: 기존 실패 흔적 정리

가장 먼저 실패한 애드온 객체를 삭제하여 깨끗한 상태로 만듭니다.

aws eks delete-addon --cluster-name $CLUSTER_NAME --addon-name aws-ebs-csi-driver

Step 2: IAM Role 생성 확인 (또는 고유 이름 사용)

이미 Role이 존재한다면 그대로 사용하고, 새로 만든다면 반드시 기존에 없는 이름을 사용해야 합니다.

• Tip: eksctl 실행 시 --role-name을 생략하면 eksctl이 고유한 이름을 자동으로 생성해 줍니다. 만약 수동 지정이 필요하다면 버전 번호나 날짜를 붙여 중복을 피하십시오.

Step 3: OVERWRITE 옵션과 함께 애드온 재설치

이미 존재하는 ServiceAccount와의 갈등을 무시하고 EKS 애드온이 관리 권한을 강제로 가져오도록 설정합니다.

aws eks create-addon \
  --cluster-name $CLUSTER_NAME \
  --addon-name aws-ebs-csi-driver \
  --service-account-role-arn arn:aws:iam::$ACCOUNT_ID:role/고유한_IAM_역할_이름 \
  --resolve-conflicts OVERWRITE

4. 최종 결과 확인

aws eks describe-addon --cluster-name $CLUSTER_NAME --addon-name aws-ebs-csi-driver --query 'addon.status'
# "ACTIVE" 확인 시 성공

5. 핵심 요약 (Lessons Learned)

1. IAM Role Name: 자동화 도구(eksctl, terraform) 사용 시 리소스 이름 중복 여부를 반드시 체크할 것.
2. Conflict Strategy: 기존에 Kubernetes 리소스(SA 등)가 존재하는 상태에서 EKS 애드온을 추가할 때는 반드시 --resolve-conflicts OVERWRITE 옵션을 고려할 것.
3. State Recovery: EKS 애드온이 CREATE_FAILED라면 업데이트는 불가능하다. 삭제(Delete) 후 재생성이 답이다.

EKS 클러스터 배포 후 kubectl get nodes를 입력했을 때 No resources found가 출력되었다. 클러스터(Control Plane)라는 뇌는 생성되었으나, 실제 연산을 수행할 워커 노드(Worker Node)가 조인되지 않은 상태다.

원인 분석

명령을 내리는 Bastion Host의 IAM 역할(Role) 부재가 도미노 현상을 일으킨 것이 핵심이다.

• 테라폼 실행 주체에게 권한이 없어 노드 그룹 생성 API 호출이 거절되었다.
• 이 과정에서 일부 리소스(KMS, Log Group)만 생성되고 상태 파일(tfstate) 기록에는 실패하여 데이터 불일치가 발생했다.

2. 주요 트러블슈팅 케이스별 대응

Case 1. AlreadyExistsException (KMS, CloudWatch)

• 증상: terraform apply 시 특정 리소스가 이미 존재한다며 중단된다.
• 이유: 이전 배포 실패로 리소스는 AWS에 생성되었으나, 테라폼의 '기억(State)'에는 등록되지 않아 발생하는 충돌이다.
• 대응: terraform import 명령어로 기존 자원을 테라폼 관리 하에 편입하거나, AWS CLI로 해당 리소스를 직접 삭제 후 재배포한다.

Case 2. failed to create iamserviceaccount (eksctl)

• 증상: AWS Load Balancer Controller 설치를 위한 서비스 계정 생성에 실패한다.
• 이유: 클러스터와 IAM 간의 신뢰 관계(OIDC Provider)가 설정되지 않았거나, 정책 ARN의 Account ID에 하이픈(-)이 포함된 경우다.
• 대응: eksctl utils associate-iam-oidc-provider를 선행 실행하고 ARN 형식을 다시 확인한다.

Case 3. Bastion Host 권한 불일치

• 증상: 모든 명령어가 AccessDenied를 반환하거나 불완전하게 실행된다.
• 이유: EC2 인스턴스 역할과 aws configure로 설정된 Access Key 권한이 충돌하여 우선순위가 정립되지 않은 상태다.
• 대응: 인스턴스 프로파일을 확인하고 ~/.aws/credentials 삭제를 통해 권한 체계를 단일화한다.

3. "Clean Slate" 전략: 완벽한 리셋 프로세스

설정이 꼬였을 때 가장 빠르게 복구할 수 있는 단계별 가이드다.

Step 1: 잔여 리소스 수동 삭제

# EKS 클러스터 삭제 (약 15분 소요)
aws eks delete-cluster --name my-eks-cluster

# KMS 및 로그 그룹 삭제
aws kms delete-alias --alias-name alias/eks/my-eks-cluster
aws logs delete-log-group --log-group-name /aws/eks/my-eks-cluster/cluster

Step 2: 테라폼 상태 초기화

# 상태 파일 및 캐시 강제 삭제
rm -rf .terraform* terraform.tfstate*

Step 3: 기반 권한 재점검

• Bastion Host에 AdministratorAccess 역할이 연결되었는지 확인한다.
• aws sts get-caller-identity로 현재 권한 주체를 확정한다.

Step 4: 재배포

terraform init
terraform plan
terraform apply -auto-approve

4. 회고

인프라 구축은 명령어의 나열이 아니라 권한(IAM) - 네트워크(VPC) - 상태(State)라는 세 요소의 정렬을 맞추는 과정이다. 이번 트러블슈팅을 통해 리소스 간의 의존성과 테라폼의 작동 원리를 깊게 이해하게 되었다.

이때 필요한 것이 Node Autoscaling이다.
대표적인 도구가 Cluster Autoscaler와 Karpenter인데, 둘 다 노드를 자동으로 늘려준다는 점은 같지만 동작 방식은 꽤 다르다.

HPA와 Node Autoscaler의 관계

• HPA: Pod 개수 조절
• Cluster Autoscaler / Karpenter: Node 개수 조절

즉, HPA가 Pod를 늘리고, Node가 부족하면 Node Autoscaler가 새 Node를 추가하는 구조다.

아키텍처 흐름

flowchart LR
    A[User Traffic Increase] --> B[HPA scales out Pods]
    B --> C[Some Pods become Pending]
    C --> D[Node Autoscaler works]
    D --> E[New Node added]
    E --> F[Pending Pods scheduled]

Cluster Autoscaler란?

Cluster Autoscaler는 미리 만들어둔 Node Group을 기준으로 확장한다. 예를 들어 AWS EKS에서 Managed Node Group이나 Auto Scaling Group을 먼저 구성해두고, 스케줄되지 못한 Pod가 생기면 그중 하나의 크기를 늘리는 방식이다.

즉, Cluster Autoscaler는 “이미 준비된 Node Group의 크기를 조절하는 방식” 이라고 볼 수 있다.

Karpenter란?

Karpenter는 Pending Pod의 요구사항을 보고, 그에 맞는 Node를 그때그때 새로 생성한다. CPU, Memory, Architecture, Spot 여부 같은 조건을 기준으로 더 적절한 인스턴스를 선택할 수 있다.

즉, Karpenter는 “Node Group 중심이 아니라 워크로드 중심으로 확장하는 방식” 이다.

두 도구의 가장 큰 차이

핵심 차이는 확장의 기준이다.

• Cluster Autoscaler: 미리 정의된 Node Group 기준
• Karpenter: Pending Pod의 실제 요구사항 기준

한 문장으로 정리하면 이렇다.

Cluster Autoscaler는 Node Group 중심, Karpenter는 워크로드 중심이다.

비교 다이어그램

flowchart TD
    subgraph CA[Cluster Autoscaler]
        A1[Pending Pod 발생] --> A2[기존 Node Group 확인]
        A2 --> A3[Node Group 크기 증가]
        A3 --> A4[새 Node 추가]
    end

    subgraph KP[Karpenter]
        B1[Pending Pod 발생] --> B2[Pod 요구사항 분석]
        B2 --> B3[적절한 인스턴스 선택]
        B3 --> B4[새 Node 생성]
    end

왜 요즘은 Karpenter를 많이 사용할까?

1. 비용 최적화

Karpenter는 Pod 요구사항에 맞는 인스턴스를 더 유연하게 선택할 수 있어서, 불필요하게 큰 Node를 유지할 가능성을 줄여준다.

2. 운영 단순화

Cluster Autoscaler는 Node Group이 많아질수록 설계와 관리가 복잡해진다. 반면 Karpenter는 Node Group 설계 부담을 줄여준다.

3. 워크로드 변화 대응

서비스가 많고 요구사항이 다양한 환경에서는, Karpenter처럼 상황에 맞는 Node를 바로 생성하는 방식이 더 잘 맞는다.

Karpenter가 선호되는 이유

flowchart LR
    A[Workload Diversity] --> D[Karpenter]
    B[Cost Optimization] --> D
    C[Operational Simplicity] --> D

Cluster Autoscaler가 여전히 괜찮은 경우

그렇다고 Cluster Autoscaler가 무조건 뒤처진 것은 아니다. 클러스터 구조가 단순하고, Node Group 수가 많지 않으며, 현재 운영 방식이 안정적이라면 Cluster Autoscaler도 충분히 좋은 선택이다.

즉, 무조건 하나가 정답이라기보다는 클러스터 규모와 운영 복잡도에 따라 선택이 달라진다.

마무리

정리하면 아래와 같다.

• Cluster Autoscaler: 미리 준비한 Node Group을 확장
• Karpenter: Pending Pod에 맞는 Node를 직접 생성

최근 EKS 환경에서 Karpenter가 많이 사용되는 이유는 비용 효율성, 운영 단순화, 유연한 확장성 때문이다.

실무에서 테라폼을 도입하면 얻을 수 있는 구체적인 장점들은 다음과 같습니다: 장애 방지 및 신속한 복구: 수동 작업 시 발생할 수 있는 실수를 방지하고, 장애 발생 시 코드를 통해 빠르게 동일한 환경을 재구축할 수 있습니다. 업무 효율성 (칼퇴 보장): 수동으로 일일이 생성하던 자원들을 코드로 단시간에 생성할 수 있어 업무 시간을 획기적으로 줄여줍니다. 컴플라이언스 대응: ISMS-P(개인정보보호법) 심사나 전자금융거래법 감사 시, 콘솔을 일일이 보여주는 대신 코드를 통해 인프라 현황을 명확하게 증명할 수 있습니다. 멀티 클라우드 및 종속성 탈피: 특정 CSP(AWS, Google Cloud, Naver 등)에 종속되지 않고 동일한 문법(HCL)으로 여러 클라우드를 관리할 수 있습니다 . 예를 들어 게임사에서 해외 데이터센터 응답 속도 최적화를 위해 AWS와 GCP를 병행 운영할 때 매우 유용합니다 . 불변 인프라(Immutable Infrastructure): 서버를 수정하는 대신 기존 것을 삭제하고 새로 생성하는 클라우드 네이티브 방식에 최적화되어 있습니다 . 잠깐! 용어 정리 CSP (Cloud Service Provider): 인프라를 제공하는 업체 (AWS, Google, Naver 등) MSP (Managed Service Provider): 기업의 클라우드 설계를 대행하고 운영/관리해주는 전문 업체 (메가존, 베스핀글로벌, 가비아 등)

2. 테라폼의 핵심 개념 테라폼을 다루기 위해 반드시 알아야 할 5가지 개념입니다 : Provider: 테라폼과 외부 서비스(AWS, GCP 등)를 연결하는 플러그인입니다. Resource: 실제 생성할 인프라 자원 (EC2, S3, VPC 등)을 정의합니다. State (.tfstate): 현재 인프라의 상태를 저장하는 파일입니다. 팀 협업 시에는 S3 같은 원격 저장소에 저장하고 Locking(DynamoDB 활용)을 하는 것이 원칙입니다. Module: 리소스들을 그룹화하여 재사용 가능하게 만든 패키지입니다. HCL (HashiCorp Configuration Language): 테라폼 전용 언어로 가독성이 매우 뛰어납니다.

3. 테라폼 동작 프로세스 테라폼은 보통 다음의 4단계 흐름으로 작업합니다 : 명령어 설명 terraform init 작업 디렉토리 초기화 및 Provider 플러그인 다운로드 terraform plan 생성/수정/삭제될 리소스 예측 및 검토 (실제 반영 X) terraform apply 실제 인프라 배포 (실행 시 yes 입력 필요) terraform destroy 생성된 모든 리소스 삭제 Tip: -auto-approve 옵션을 사용하면 yes 입력 없이 바로 실행이 가능합니다.

4. 실전! 테라폼으로 EKS(Elastic Kubernetes Service) 구축하기 비용 절감 꿀팁 EKS Managed Node Group 설정 시 비용을 아끼고 싶다면 capacity_type을 ON_DEMAND가 아닌 SPOT으로 설정하세요 . 주요 구축 흐름 VPC 생성: EKS를 위해 최소 2개의 가용 영역(AZ)과 Public/Private 서브넷을 구성합니다. EKS 클러스터 생성: 공식 EKS 모듈을 사용하여 Control Plane과 노드 그룹을 설정합니다. 접속 설정: 인프라 생성 후 내 PC에서 클러스터를 확인하려면 kubeconfig 업데이트가 필요합니다.

5. 운영 중 발생하는 문제 및 꿀팁 (Troubleshooting) Q1. HPA 설정 후 서버 부하를 주었는데 파드가 늘어나지 않아요. 원인: Metrics Server가 설치되어 있는지 확인하세요. 작동 원리: HPA 컨트롤러는 15초마다 메트릭 서버를 체크하여 목표 사용량과 대조 후 Pod 개수를 조절합니다. Q2. 노드 자리가 부족해서 노드를 늘려야 해요. Cluster Autoscaling이 필요하며, 이때 ASG(Auto Scaling Group) 이름과 클러스터 이름 두 가지 정보가 필수입니다. Q3. 재설치 시 CloudWatch Logs 그룹이 이미 존재한다는 오류가 떠요. 이 경우 다음 순서로 조치하세요 : 콘솔에서 EKS 삭제 CloudWatch Logs 그룹 수동 삭제 로그 그룹 재사용 선언 후 재설치

6. 마치며 테라폼은 이제 클라우드 엔지니어에게 선택이 아닌 필수 도구입니다. 더 많은 프로바이더 사용 예시는 Terraform Registry에서 직접 확인하고 적용해 보시기 바랍니다.

#Terraform #IaC #AWS #EKS #CloudNative #Kubernetes #멀티클라우드