운영·스테이징 간 스트리밍 체크포인트 충돌

감사 이벤트를 Kafka에서 Delta 테이블로 가져오는 파이프라인을 만들고 있었습니다. 과거 로그를 옮기는 배치와 새 이벤트를 읽는 스트림이 같은 분석 테이블을 사용하는 구성이었습니다.

테이블 스키마와 집계 기준을 맞추던 중 운영 작업이 테스트 토픽을 읽으려는 오류가 발생했습니다. 병렬 실행에서도 충돌이 나타나 처음에는 권한과 저장소 경합을 의심했습니다. 원인은 실행 환경을 바꾸기 전에 가져온 Python 설정 변수와 공유된 체크포인트였습니다.

테이블 저장 방식

기존 배치가 만든 테이블에는 새 스트림에 없는 컬럼이 있었습니다. 모든 컬럼을 암묵적으로 넣는 방식에서 쓰기가 실패해, 적재할 컬럼을 명시하도록 바꿨습니다.

이벤트별 물리 테이블을 계속 따로 관리하는 대신 통합 Silver 테이블 위에 View를 두었습니다. 배치와 스트림이 같은 정제 규칙을 사용하도록 하고, 집계에서 쓰는 이벤트명과 조인 키도 맞췄습니다. 작업은 끝났는데 집계가 비어 있을 때는 실제 데이터에 적용된 필터까지 확인했습니다.

운영 환경의 테스트 토픽 권한 오류

운영 작업이 테스트 토픽에 접근하려는 오류가 나왔을 때는 실행 환경 설정을 빠뜨린 것으로 생각했습니다. 체크포인트를 정리해 임시로 복구했지만 병렬 실행 충돌은 다시 나타났습니다.

저장소 잠금 경합을 의심해 순차로 실행하고 테스트와 운영의 시간도 나눴습니다. 오류를 줄이기 위한 우회였지만 이 설정을 바꾼다고 원인이 없어지는 것은 아니었습니다.

다시 로그를 보면서 테스트 작업이 운영 체크포인트 경로를 사용하고 있다는 것을 찾았습니다. 코드에는 환경별 경로를 지정하는 함수가 있었지만, 실행할 때 실제로 읽는 값은 다른 상태였습니다.

Python 설정 변수의 바인딩 문제

노트북은 환경 설정 함수를 실행하기 전에 체크포인트 변수를 가져오고 있었습니다. 아래는 이 문제를 설명하는 간단한 예시입니다.

checkpoint_base = "checkpoints/production"
 
def configure(environment):
    global checkpoint_base
    checkpoint_base = f"checkpoints/{environment}"

from settings import checkpoint_base, configure
 
configure("staging")
print(checkpoint_base)  # checkpoints/production

from settings import checkpoint_base는 그 시점에 가리키는 객체를 현재 모듈의 이름에 연결합니다. 이후 configure()에서 settings.checkpoint_base를 다른 문자열로 재할당해도 이미 가져온 이름까지 갱신되지는 않습니다. 객체를 깊은 복사해서 생기는 문제는 아닙니다.

실제 작업에서는 설정 적용 후 값을 가져오도록 순서를 고쳤습니다. 같은 문제를 피하는 다른 방법으로는 모듈을 import하고 설정 후 속성을 읽는 방식이 있습니다.

import settings
 
settings.configure("staging")
print(settings.checkpoint_base)  # checkpoints/staging

실행 직전의 경로 확인

환경별 설정 함수가 있다는 것과 실제 스트림에 전달된 경로가 다르다는 것이 이번 문제의 핵심이었습니다. 따라서 수정 후에는 설정 파일을 읽는 데서 끝내지 않고 스트림 시작 직전에 환경, 대상 테이블, 체크포인트를 한 묶음으로 확인했습니다.

다음은 환경별 구성을 불변 값으로 전달하는 대안 예시입니다. 당시 적용한 코드를 그대로 옮긴 것은 아니고, 전역 변수의 재할당에 의존하지 않는 형태를 보여줍니다.

from dataclasses import dataclass
 
@dataclass(frozen=True)
class StreamConfig:
    environment: str
    target: str
    checkpoint: str
 
def config_for(environment, stream_name):
    if environment not in {"production", "staging"}:
        raise ValueError("unknown environment")
    return StreamConfig(
        environment=environment,
        target=f"{environment}.silver.{stream_name}",
        checkpoint=f"checkpoints/{environment}/{stream_name}",
    )
 
prod = config_for("production", "audit")
stage = config_for("staging", "audit")
assert prod.checkpoint != stage.checkpoint

이 예시는 경로의 구분을 확인하는 코드입니다. 실제 환경에서는 저장소의 절대 경로와 권한, 쿼리별 디렉터리까지 맞아야 합니다. 환경 이름만 바뀌고 두 설정이 같은 저장 위치를 가리키면 같은 문제가 생길 수 있습니다. 대상 테이블을 나눴다고 체크포인트까지 자동으로 분리되는 것도 아닙니다.

바인딩 문제를 작은 코드로 확인

Spark를 실행하지 않고도 Python의 이름 연결을 재현할 수 있습니다. 다음 코드는 앞에서 설명한 문제와 모듈 속성을 읽는 차이를 확인합니다.

from types import SimpleNamespace
 
settings = SimpleNamespace(checkpoint_base="checkpoints/production")
imported_name = settings.checkpoint_base
settings.checkpoint_base = "checkpoints/staging"
assert imported_name == "checkpoints/production"
assert settings.checkpoint_base == "checkpoints/staging"

여기서 바뀌는 것은 문자열 객체의 내용이 아니라 settings.checkpoint_base가 연결하는 대상입니다. 노트북에서 이미 실행한 셀과 다시 실행한 셀이 섞이면 설정 적용 순서도 놓치기 쉽습니다. 셀의 화면상 순서보다 실제 시작 코드가 어떤 값을 읽는지 확인해야 했습니다.

운영 로그에는 비밀값 대신 환경과 논리 대상, 체크포인트 식별 정보를 남기면 비교할 수 있습니다. 권한 오류의 토픽 이름만 보고 운영 계정에 테스트 권한을 더 주는 방향으로 해결하지 않은 이유도 같습니다. 접근 권한이 필요한 요청인지부터 다시 확인해야 했습니다.

수정 후 병렬 실행 복구

환경 설정에 분기가 있어도 실제 체크포인트 경로가 같으면 두 작업은 같은 진행 기록을 사용하게 됩니다. 제 경우에도 테스트와 운영이 서로의 기록을 건드리고 있었습니다.

경로를 분리한 뒤에는 순차 실행과 실행 시간 분리 우회를 제거하고 병렬 실행을 복원했습니다.

설정 함수에 적힌 경로보다 실행 로그의 실제 경로가 원인을 좁히는 데 도움이 됐습니다. 이후에는 환경 이름뿐 아니라 실제 테이블과 체크포인트 경로도 함께 확인하게 됐습니다.

다음 글: 스트리밍 수집 건수와 체크포인트 복구 오류

참고 자료

• Python import와 이름 바인딩 
• Spark Structured Streaming과 Kafka