소개
스튜디오는 원래 표시 애플리케이션이 일시 중지된 상태에서 투표 이력을 보관하고 있습니다. 애플리케이션의 진행 상황을 되돌리거나 로그를 바꾸지 않고 새 집계를 만들어야 합니다. 새 컨슈머가 고정된 배치를 아직 받지 못했어도 폴링에서 잠시 메시지가 반환되지 않을 수 있습니다.
네이티브 리플레이 루프를 만들고 각 파티션의 끝 위치를 따로 확보한 뒤, 유한한 전체 배치를 읽은 후에만 결과를 내보냅니다. 이전 실습의 논리적 식별 함수는 이 독립적인 VM에 이미 제공됩니다. 거기서 사용한 Python과 Kafka 기초가 필요합니다. Terminal을 사용하세요. 작성자의 50~60분 예상에는 읽기와 일반적인 오류 처리가 포함되며, 초보자 대상 시간 측정 실험은 아닙니다.
이는 고정된 배치를 재구성하는 작업입니다. 실습 중 이 토픽들에 메시지가 계속 추가되지 않으며, 구현해야 할 트랜잭션도 없습니다. 신뢰할 수 있는 비정상 종료 후 복구나 종단 간 정확히 한 번 처리를 보장하지 않습니다.
각 파티션의 경계를 확보하고 경계까지 읽기
이 단계에서는 준비된 바깥 프로그램 안에 완전히 독립적인 네이티브 읽기를 구현합니다. Terminal에서 시작하세요. replay.py와 필요한 경우 reducer.py 또는 helpers/ 아래의 Python 보조 프로그램을 편집하고, 파생 파일은 output/ 아래에 기록하세요. 제공된 도구/데이터, 원래 토픽, 일시 중지된 표시 애플리케이션의 그룹을 보존합니다.
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
읽기 전용 설명에는 실제 studio-votes 파티션 ID, 보존된 시작점 low, 현재 고정된 끝, 레플리카가 표시됩니다. 파티션은 순서가 있는 로그이지 별도 VM이 아닙니다. 끝은 보존된 배치 뒤의 다음 위치이며 세어야 할 레코드가 아닙니다. 어느 파티션에서 레코드가 필요한지 예상해 보세요. 빈 파티션은 low == high이므로 이미 완료되어 있습니다.
오프셋 범위의 폭은 레코드 개수가 아닙니다. Kafka는 내부/제어 위치나 삭제된 위치를 건너뛸 수 있습니다. 이 원본에는 실제로 건너뛴 네이티브 위치가 있습니다. 만들어 넣어야 할 누락된 표는 없습니다. 실제 레코드 전체 정보를 확인하여 실제 위치와 설명을 비교하세요.
bigdata-python -B vote_tool.py inspect main
재구성 전에 선수 과정의 컨슈머 그룹 도구로 일시 중지된 표시 애플리케이션의 저장 위치를 읽습니다.
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
나중에 비교할 수 있도록 각 CURRENT-OFFSET을 보관하세요. 활성 멤버가 없다는 메시지는 정상입니다. 표시 애플리케이션은 일시 중지되어도 저장된 진행 상황은 남습니다. 여기서 LAG는 로그 위치의 차이이며 논리적인 표 수가 아닙니다. 독립 컨슈머는 이 그룹의 진행 상황을 움직이면 안 됩니다.
준비된 바깥 프로그램 읽기
cat replay.py
import, arguments(), 클라이언트 설정, 오류 보고, finally/닫기는 이미 제공됩니다. 핵심 부분은 의도적으로 NotImplementedError를 발생시키며 파티션을 읽거나 예제 답을 내보내지 않습니다. 이전 실습의 올바른 식별 함수는 reducer.py에 따로 준비되어 있습니다.
클라이언트는 선택한 bootstrap.servers 주소와 독립적인 group.id를 사용합니다. allow.auto.create.topics는 false입니다. 자동 커밋과 자동 오프셋 저장은 꺼져 있습니다. assign은 동적 그룹 할당 없이 명시적인 시작 위치를 부여하며 이 예제는 명시적인 커밋을 호출하지 않습니다. 동등한 독립 그룹 구현은 자신의 진행 상황을 저장해도 되지만 원래 표시 애플리케이션 그룹에 가입하거나, 커밋하거나, 초기화하거나, 삭제해서는 안 됩니다.
준비된 vote_support.py는 토픽/카탈로그/출력/broker/그룹용 arguments()와 완전한 결과를 기록하는 finish(...)를 제공합니다. 파티션을 발견하거나 투표를 가져오는 함수는 아닙니다. finish는 결과 쓰기/교체 전에 식별 규칙을 호출합니다. finally는 성공 또는 실패 시 컨슈머를 닫습니다. 이 바깥 구조는 유지하세요.
짧게 이어지는 부분들로 네이티브 핵심 구현하기
이전 실습의 편집 흐름을 사용하세요.
nano replay.py
import, 인수 해석, Consumer 설정, try:, except, finally 부분은 유지합니다. try: 안에서 설명 주석 두 개와 미완성 핵심 부분의 raise NotImplementedError(...)만 제거하세요. 화살표 키로 위치를 잡고 Backspace/Delete로 편집합니다. Ctrl+K는 현재 행 전체를 잘라 낼 수 있으므로 교체하려는 행에 있을 때만 사용하세요.
그 빈자리에 다음의 연결된 부분들을 넣습니다. 이들은 하나의 핵심 부분입니다. 개별 블록을 셸에서 실행하거나 반쯤 만든 반복문을 저장/실행하지 마세요. 모든 부분을 완성하고 제공된 바깥 오류/닫기 틀을 유지한 다음 저장하고 실행합니다. 공백 네 개는 문장을 try: 안에 놓고 네 개를 더하면 반복문이나 조건문 안에 놓습니다. 표시된 상대적 들여쓰기는 Python의 의미에 포함됩니다.
먼저 실제 전체 파티션 집합을 발견하세요. 토픽을 지정하지 않고 list_topics를 호출하면 알 수 없는 이름에 대한 메타데이터를 요청하지 않습니다. 선택한 항목을 사용하기 전에 사용 가능한지 확인합니다. 발견한 ID를 정렬하면 이 예제의 할당 목록이 일정해지지만 전역 이벤트 순서가 생기는 것은 아닙니다.
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
발견한 모든 파티션에 대해 get_watermark_offsets(..., timeout=10)를 한 번씩 요청합니다. TopicPartition은 그 요청의 토픽과 파티션을 식별합니다. 보존된 low와 고정된 high를 모두 보관하고 진단용 다음 위치를 low로 초기화하세요. 폴링 루프 안에서 high를 갱신하지 마세요. 커지는 끝을 쫓으면 더 이상 유한한 재집계가 아닙니다. 이 실습은 고정된 원본에 추가하거나 변경하는 일을 금지합니다.
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
읽을 보존 위치가 있는 파티션만으로 pending을 구성합니다. assign은 빈 파티션의 low도 포함하여 명시적인 시작 오프셋 목록을 받습니다. 실행마다 새 물리 레코드 목록을 초기화하세요. time.monotonic()은 시스템 시계 조정과 무관하게 경과 시간을 측정합니다. 마감 시점은 미완료 읽기에 제한을 둡니다.
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
이제 반복문을 시작합니다. poll(0.2)는 메시지, 오류/이벤트가 있는 메시지 또는 시간 초과 시 None을 반환할 수 있습니다. 왜 None이 완료를 뜻할 수 없는지 예상해 보세요. 클라이언트가 고정된 레코드를 아직 받지 못했을 뿐일 수 있습니다. 전체 마감 시점 안에서 계속 기다립니다. 설정에서 파티션 EOF 이벤트가 켜져 있습니다. _PARTITION_EOF는 예상한 끝 이벤트이고 다른 오류는 실행 실패입니다. 이를 통해 마지막 내부 위치를 넘어선 네이티브 진행도 처리할 수 있습니다.
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
같은 if message is not None 블록을 다음 elif로 이어 가세요. 공백 열두 개로 if message.error():와 나란히 맞춥니다. 미완료 파티션의 정상 메시지 중 고정된 high보다 낮은 위치의 메시지만 레코드 전체 정보로 담습니다. 물리적인 재전송도 보관하세요. summarize가 나중에 논리적 식별을 처리합니다. UTF-8 키/값 바이트를 디코딩하고 json.loads로 값을 딕셔너리로 바꿉니다.
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
이번 폴링을 처리한 뒤 남은 모든 파티션에 대해 consumer.position(...)을 조회합니다. 이는 컨슈머의 다음 위치이며 어떤 그룹의 커밋된 진행 상황과도 다릅니다. 추측한 레코드 개수가 아니라 확보한 high와 각각 비교하세요. 완료된 파티션만 pending에서 제거합니다. for position 행은 공백 여덟 개로 while 안이면서 메시지 조건 밖에 놓습니다. 따라서 빈 폴링 후에도 위치를 검사할 수 있습니다. 마지막으로 반복문이 완료된 뒤에, 공백 네 개로 finish를 호출하세요.
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
이제 각 파티션에는 고정된 경계가 있습니다. 빈 파티션에는 표가 필요 없고, 건너뛴 위치에는 가짜 레코드가 필요 없으며, None이 읽기를 멈추지 않습니다. 모든 미완료 파티션의 다음 위치가 각 high에 도달한 후에만 완전한 스냅샷을 내보냅니다.
Ctrl+O, Enter로 replay.py를 확인한 다음 Ctrl+X로 저장하고 종료하세요. 저장된 프로그램을 살펴봅니다.
cat replay.py
여섯 부분이 모두 제공된 try 안에 있는지, 레코드 전체 정보 처리가 정상 메시지 분기 안에 있는지, 위치 검사가 반복문 안에 있는지, finish는 그 밖에 있는지 확인하세요. 이어서 바로 선택한 네이티브 배치에 실행합니다.
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
물리 레코드 여덟 개, 논리적인 다섯 표, cedar=3, maple=2, willow=0이 보여야 합니다. 공개 진입점은 먼저 식별 함수를 일회용 예제로 검사한 뒤 실제 네이티브 프로그램/보조 프로그램 트리가 받은 broker 데이터를 기록합니다. 별도의 완전한 비즈니스 비교가 실제 결과를 검사합니다. 스냅샷이 보고한 경계/위치는 진단용이며 네이티브 처리를 독립적으로 증명하지는 않습니다.
성공한 각 호출은 결과를 실제로 기록해야 합니다. 직접 쓰거나 임시 파일에 쓴 뒤 선택한 출력으로 이름을 바꿔도 됩니다. 반복 실행에서 같은 바이트도 유효하고 보조 자식 프로세스가 써도 됩니다. 옛 파일을 유지하거나 타임스탬프만 갱신하거나 완료 마커를 쓰는 것은 결과 기록이 아닙니다. 제공된 finish는 배운 임시 파일/교체 방식을 사용합니다. 준비된 이 기록 방식을 사용하면 결과 기록 이전의 루프나 검증 오류가 이전 스냅샷을 사용할 수 있게 남겨 둡니다. 진입점이 임의의 제자리 쓰기를 롤백하는 것은 아닙니다.
문법/이름/들여쓰기 오류라면 보고된 행을 따라 nano replay.py를 다시 열어 해당 부분을 고치고 작동하는 부분은 유지하세요. 저장/종료하고 cat replay.py로 확인한 뒤 공개 진입점을 다시 실행합니다. 미완료 리플레이 시간 초과에는 선택한 토픽 인수와 pending, EOF, 위치 처리를 확인하세요. 원래 그룹을 되감을 이유는 아닙니다. pwd와 ls replay.py reducer.py로 잘못 입력한 경로를 찾으세요. 로컬 프로그램 오류 때문에 setup을 다시 실행하거나 원본 레코드를 바꿀 필요는 없습니다.
처음부터 재구성하고 다른 배치 사용하기
이 단계에서는 다시 실행해도 같은 유한한 결과를 재구성함을 확인한 다음, 새로운 카탈로그와 다른 파티션 분포를 사용합니다. Terminal의 /home/labex/project에서 계속 작업하세요.
준비된 기록 방식은 JSON을 저장하기 전에 물리 레코드 전체 정보를 파티션/오프셋 순서로, 논리적인 표를 ID 순서로 정렬합니다. 이를 통해 이벤트의 전역 순서를 보장하지 않고도 동일한 완전한 스냅샷을 안정적으로 만들 수 있습니다. sha256sum은 내용의 지문을 계산합니다. >는 지문을 출력 디렉터리에 저장하며, 나중에 -c가 실제 지정 파일을 저장된 지문과 비교합니다.
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
output/main-snapshot.json: OK가 보여야 합니다. 프로그램은 매 호출마다 비어 있는 records, seen, 집계에서 시작합니다. 이력의 표를 이전 합계에 또 더하지 않고 완전한 스냅샷을 교체합니다. 소비가 레코드를 삭제하거나 일시 중지된 표시 애플리케이션을 되감은 것은 아닙니다.
반복 실행 후 원래 표시 애플리케이션의 위치를 다시 읽으세요.
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
모든 CURRENT-OFFSET이 이전에 기록한 일시 중지 애플리케이션의 표와 같아야 합니다. 독립 재구성은 보존된 레코드를 읽었어도 그 그룹에 커밋하거나 되돌리지 않았습니다. 활성 멤버가 없다는 메시지도 여전히 정상입니다.
check 배치를 나열하고 자체 부스 카탈로그를 확인하세요.
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
이 배치는 부스 이름이 다르며 파티션도 네 개 대신 세 개입니다. 동일한 프로그램 파일을 유지하세요. 인수 해석기가 실제 토픽과 카탈로그를 선택하고, 발견 과정은 실제 전체 파티션 집합을 찾습니다.
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
물리 레코드 여섯 개, 논리적인 네 표, amber=2, violet=2, moss=0이 보여야 합니다. 첫 카탈로그를 하드코딩하거나 첫 토픽의 레코드 수에서 멈추면 이 배치를 올바르게 설명할 수 없습니다.
첫 번째 실제 파일로 돌아갑니다.
bigdata-python -B read_snapshot.py output/main-snapshot.json
스튜디오의 결과가 여전히 들어 있습니다. 읽기 도구는 요청한 파일을 직접 열고 Kafka에서 아무것도 보충하지 않습니다. 파일이 없다고 하면 vote_tool.py list와 파일 이름을 비교해 경로를 수정하세요. 해당 결과가 정말 재구성을 필요로 할 때만 그 배치를 다시 만드세요.
충돌을 거부하면서 유효한 스냅샷 유지하기
이 단계에서는 유용한 결과를 교체하지 않고 데이터 오류를 관찰합니다. Terminal의 /home/labex/project에서 계속 작업하세요. 목록의 conflict 토픽에는 하나의 ID에 서로 다른 유효한 부스 내용이 의도적으로 들어 있습니다. 출력 대상은 기존 주 스냅샷입니다. 어느 값도 정답으로 허용되지 않으므로 이 실행은 성공 결과로 내보내면 안 됩니다.
이전 단계의 지문은 유효한 주 파일을 설명합니다. 같은 프로그램을 충돌 배치에 실행하세요. 다음 명령은 실패해야 정상입니다.
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
0이 아닌 종료 상태와 실제 ID를 식별하는 Conflicting event_id가 보여야 합니다. 규칙은 준비된 기록 방식이 결과를 생성/교체하기 전에 실행됩니다. 여기서 .pending/교체 방식은 완전한 결과를 내보내는 기법이지, 비정상 종료 후 복구나 내구성을 보장하지 않습니다.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
유효한 주 파일은 스튜디오의 원래 집계를 유지하며 지문과 계속 일치해야 합니다. 실패한 충돌 실행은 성공한 main/check 실행과 별도로 기록됩니다. 충돌 원본을 편집하거나 먼저 도착한 이벤트를 임의로 선택하지 마세요. 역순 또는 파티션별 순차 탐색도 같은 충돌을 거부해야 합니다.
단순한 배치 이름 오타라면 진입점은 리플레이 전에 가능한 이름을 출력합니다. vote_tool.py list로 확인한 뒤 올바른 이름을 다시 실행하세요. 로컬 프로그램 오류라면 파일을 고치고 영향을 받은 유효한 배치를 다시 실행합니다. 소스 프로그램을 어떤 방식으로든 편집한 후에는 두 유효 배치를 모두 다시 실행해 결과가 현재 프로그램 파일에 속하게 한 다음, 충돌 자체 검사를 반복하세요.
원래 표시 애플리케이션의 그룹은 일시 중지 상태로 유지하세요. 가입, 커밋, 초기화, 삭제는 허용되지 않습니다. 초기화한 뒤 같은 최종 오프셋으로 돌려놓아도 독립 리플레이의 범위 밖입니다. 기능 검사는 프로그램 트리의 네이티브 수신과 실제 전체 출력을 함께 확인하지만, 계산의 인과관계나 관리자가 환경을 위조하는 데 대한 내성을 증명하지 않습니다.
요약
각 파티션의 실제 경계를 확보하고, 일시적인 빈 폴링, 빈 파티션, 건너뛴 위치가 있어도 전체 유한 로그를 읽었습니다. 동일한 네이티브 프로그램은 이전 합계를 누적하지 않고 두 카탈로그의 결과를 재구성했습니다. 충돌하는 ID는 실패했지만 유용한 주 스냅샷과 원래 컨슈머 진행 상황은 보존되었습니다.
