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

테스트와 운영의 체크포인트를 분리한 뒤에는 수집 건수 집계와 중단된 스트림의 복구 방식을 살펴봤습니다.

로그의 숫자를 실제 수집량으로 믿기 어려운 상황이 있었습니다. 여러 스트림이 같은 테이블에 쓰고 있었는데 전체 테이블의 증가량을 개별 스트림의 결과로 기록하고 있었습니다. 이를 고치는 과정에서는 최근 진행 정보의 보관 범위 때문에 실제 데이터가 들어와도 0건이나 일부 건수만 로그에 남는 문제를 만났습니다.

별도로 체크포인트 복구 로직에서도 정상 스트림을 오래된 상태로 오인하는 문제가 있었습니다. 두 문제 모두 어떤 값을 현재 실행 상태로 읽고 있는지 다시 확인해야 했습니다.

처리 건수 집계 방식 변경

먼저 recentProgress에 있는 입력 건수를 합산하는 방식으로 바꿨습니다. 하지만 이 속성은 쿼리의 전체 이력을 반환하지 않습니다. Spark가 보관하는 최근 진행 정보만 반환하며, 보관 개수에도 설정값이 있습니다.

따라서 배치가 많이 실행된 뒤 마지막에 합산하면 앞부분의 입력 건수가 빠질 수 있습니다. 실제로는 데이터가 들어왔는데 로그의 숫자는 전체 처리량과 달라지는 이유였습니다.

당시에는 스트림을 식별하는 조건으로 대상 테이블을 걸러 실행 전후 건수를 비교하도록 다시 바꿨습니다.

마지막 방식도 실행별 입력 건수를 일반적으로 보장하는 방법은 아닙니다. 같은 대상에 다른 작업이 쓰거나 중복 제거·갱신이 포함되면 테이블 증가량과 입력 건수는 달라질 수 있습니다. 어떤 숫자를 남길지와 그 숫자가 의미하는 범위를 함께 정해야 합니다.

오프셋 복구 과정의 중복 수집

스트림이 Kafka 보관 기간보다 오래 중단되면 체크포인트가 가리키는 원본 데이터가 사라질 수 있습니다. 이를 감지해 다시 읽는 처리를 넣었는데, 정상 실행에서도 복구가 반복됐습니다.

원인은 현재 진행 위치를 읽는 코드였습니다. sources/0/에 남은 시작 오프셋을 마지막으로 읽은 위치로 잘못 해석하고 있었습니다. 실행이 진행돼도 시작값은 그대로여서 계속 오래된 상태라고 판단한 것입니다.

당시 확인한 체크포인트 구조를 단순화하면 다음과 같습니다. 실제 파일 형식과 세부 구조는 실행 환경에 따라 확인해야 합니다.

checkpoint/
  sources/       # 소스의 초기 정보
  offsets/       # 배치별 오프셋 기록
  commits/       # 완료된 배치 기록

진행 기록을 읽는 위치와 배치 번호의 정렬 방식을 수정했습니다. 파일명 9와 10을 문자열로 정렬하면 순서가 달라질 수 있어 숫자로 비교해야 했습니다.

하나의 최댓값으로 비교할 수 없는 이유

Kafka 파티션의 오프셋은 서로 독립된 위치입니다. 한 파티션의 숫자가 크다고 다른 파티션의 누락이 사라지지는 않습니다. 아래는 문제가 드러나는 가상 예시입니다. 저장된 값은 다음에 읽을 위치라고 가정합니다.

saved_next = {("audit", 0): 1000, ("audit", 1): 10}
earliest = {("audit", 0): 900, ("audit", 1): 20}
 
# 최댓값만 비교하면 파티션 1의 누락을 놓칩니다.
assert max(saved_next.values()) >= max(earliest.values())
gaps = {
    key: (saved_next[key], first_available)
    for key, first_available in earliest.items()
    if key in saved_next and saved_next[key] < first_available
}
assert gaps == {("audit", 1): (10, 20)}

0번 파티션은 이어 읽을 위치가 남아 있지만 1번 파티션의 10~19 구간은 현재 원본에서 읽을 수 없는 상황입니다. 실제 판단에는 토픽·파티션 대응뿐 아니라 완료된 배치의 위치인지, 오프셋이 시작·끝 중 무엇을 뜻하는지도 맞춰야 합니다. 위 코드는 수치 비교의 문제를 보여주는 예제이며 Spark 체크포인트를 수정하는 복구 도구가 아닙니다.

신규 파티션이 있거나 저장된 파티션을 현재 조회 결과에서 찾지 못한 경우도 단순히 “누락 없음”으로 넘길 수는 없습니다. 보관 정책과 토픽 변화, 조회 오류를 구분할 정보가 필요합니다. 누락 여부를 확인하지 못한 경우에는 정상으로 처리하지 않도록 했습니다.

입력 건수와 저장 건수의 비교

최근 진행 기록을 합산하는 문제도 작은 목록으로 설명할 수 있습니다.

all_batches = [100, 200, 300, 400]
retained_recent_batches = all_batches[-2:]
assert sum(all_batches) == 1000
assert sum(retained_recent_batches) == 700

실제 recentProgress의 보관 개수와 이 예시의 2개는 관계가 없습니다. Spark 문서 에서 설명하는 최근 진행 정보가 어떤 값인지 보여주는 예시입니다. 최근 값이 모두 정상이어도 앞부분의 기록까지 포함한 누계는 아닐 수 있습니다.

반면 대상 테이블 증가량은 입력 건수와 다른 숫자입니다. 중복 제거로 100건을 받아 80건만 새로 저장할 수도 있고, 갱신 작업이면 행 수가 그대로일 수도 있습니다. 로그에는 읽은 건수, 새로 저장된 건수, 현재 테이블 행 수 중 무엇을 남겼는지 이름을 붙이는 것이 필요했습니다.

체크포인트 복구 시 주의할 점

원래 수정 코드에는 파티션별 오프셋을 하나의 최댓값으로 요약하는 부분도 있었습니다. 하지만 Kafka 오프셋은 토픽과 파티션별 위치입니다. 서로 다른 파티션의 숫자를 하나로 합쳐 비교하면 어느 파티션에 데이터가 남아 있는지 판단할 수 없습니다.

또한 offsets에 기록이 있다는 것만으로 해당 배치의 쓰기까지 끝났다고 볼 수는 없습니다. 완료 기록과 파티션별 위치를 함께 확인해야 합니다. 당시 코드를 그대로 범용 복구 예제로 제시하기 어려운 이유입니다.

체크포인트를 새로 만들면 이전 데이터를 다시 읽어 중복이 생길 수 있습니다. 반대로 보관 기간이 지나 삭제된 원본은 체크포인트를 바꿔도 돌아오지 않습니다. 처리 건수가 0이라는 이유만으로 진행 기록을 삭제해서는 안 됩니다.

당시 Bronze에 쌓인 중복을 정리하고 후속 정제·집계 경로도 확인했습니다. 시작 오프셋을 현재 처리 위치로 잘못 읽은 판단이 복구 작업으로 이어진 문제였습니다. 이후에는 처리 건수와 파티션별 진행 위치를 구분하고, 실제 누락을 확인한 뒤 복구 여부를 결정하는 쪽으로 정리했습니다. 체크포인트를 새로 만드는 작업도 중복이나 원본 유실을 해결해 주는 만능 복구 방법으로 두지 않았습니다.

참고 자료

• PySpark StreamingQuery.recentProgress 
• Spark Kafka 오프셋과 데이터 손실 옵션