소개
작은 갤러리에서 방문객은 가장 좋아하는 부스에 투표합니다. 송신자가 같은 이벤트를 재전송할 수 있으므로 새 Kafka 레코드가 항상 새 표를 뜻하지는 않습니다. 먼저 정상적인 두 표 예제를 살펴보고, 실제로 다른 저장 위치에 기록된 재전송을 관찰한 다음, 갤러리의 투표 수를 재구성합니다.
터미널 경로, Python 함수/반복문/딕셔너리, JSON, Apache Kafka for Beginners에서 다루는 Kafka 키, 파티션, 오프셋, 컨슈머 그룹, Python 클라이언트 기초를 알고 있어야 합니다. 이 실습은 비즈니스 이벤트의 식별 기준을 추가합니다. 집계 규칙에 집중할 수 있도록 완전한 네이티브 읽기 프로그램이 준비되어 있으며, 다음 가이드 실습에서 이 읽기 프로그램을 만듭니다. Terminal과 설치된 bigdata-python 환경을 사용하세요. 작성자의 40~50분 예상에는 읽기와 일반적인 실수 처리가 포함되며, 초보자 실측 시간은 아닙니다.
이미 전달된 유한한 배치만 다룹니다. 투표 자격 확인 시스템, 지속적인 스트림, 비정상 종료 후 복구 보장, 종단 간 정확히 한 번 처리 보장은 없습니다. 각 실습은 독립적으로 시작합니다.
다른 저장 위치에 기록된 재전송 관찰하기
이 단계에서는 정상적인 두 표와 재전송된 이벤트를 비교합니다. Terminal에서 시작하세요. 작업 공간에는 준비된 도구와 카탈로그가 있습니다. reducer.py, replay.py, 필요한 경우 helpers/ 아래의 Python 보조 프로그램만 수정하고, 파생 파일은 output/ 아래에 두세요. 제공된 도구/데이터, 고정된 원본 토픽, 일시 중지된 표시 애플리케이션의 그룹을 보존하세요. 별도의 데모 토픽에서 허용된 동작은 아래 재전송 하나뿐입니다.
cd /home/labex/project
-B는 이 Python 명령들이 바이트코드 캐시 파일을 만들지 않게 합니다. 읽기 전용 inspect 도구는 실제 broker 레코드의 전체 정보를 출력합니다. 여기에는 토픽, 파티션, 오프셋, 키, 디코딩한 값이 포함됩니다. 마지막 개수는 저장된 애플리케이션 레코드 수입니다. 작은 데모를 살펴보세요.
bigdata-python -B vote_tool.py inspect demo
두 레코드가 보여야 합니다. 값의 스키마는 다음과 같습니다. event_id, booth, choice는 비어 있지 않은 문자열이고, choice는 정확히 favourite이며, booth는 선택한 카탈로그에 있어야 합니다. 이 과정의 투표 스키마에는 추가 필드가 없습니다. 키는 라우팅에 사용하는 부스 이름입니다. event_id는 비즈니스상 한 표를 식별합니다. 여기서는 demo-1이 oak에, demo-2가 fern에 투표합니다. 두 레코드이자 두 개의 논리적인 표입니다.
demo-1을 여전히 비즈니스상 한 표로 본다면 전송 후 물리 레코드는 몇 개가 되고 부스별 집계는 어떻게 될지 먼저 예상해 보세요. 그런 다음 예상이 맞는지 시험합니다.
재전송 명령은 완전히 같은 demo-1 값을 실제 broker로 보내고 전달 확인을 기다립니다. 보내기 전에 기존 데모를 검사하므로 예상된 재전송이 이미 있는 상태에서 다시 실행해도 네 번째 레코드를 추가하지 않습니다.
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
이제 레코드는 세 개입니다. demo-1이 서로 다른 두 오프셋에 나타나며 값의 내용은 같습니다. 이 위치들은 저장된 두 레코드를 식별하지만, 반복된 이벤트 ID가 식별하는 표는 여전히 하나입니다. 올바른 부스별 합계는 oak=1, fern=1, reed=0 그대로입니다. 기존 레코드를 다시 읽는 것과 재전송을 추가로 저장하는 것은 다른 상황입니다. 둘 다 집계를 만들 때 비즈니스 식별 규칙이 필요합니다.
고정된 과정 배치는 파티션을 명시적으로 지정해 배치하므로 같은 부스 키가 서로 다른 파티션에 나타날 수 있습니다. 레코드를 비교할 때 실제 메타데이터를 읽으세요.
선택한 과정 배치 안에서 이벤트 ID는 한 파티션이나 부스에만 국한되지 않고 모든 파티션에 걸쳐 전역적으로 적용됩니다. 다른 모든 필드가 같아도 서로 다른 두 ID는 각각 한 표로 셉니다. 제공된 비즈니스 규칙은 그런 이벤트를 허용하며, 실제 사람이 몇 표를 던져도 되는지는 결정하지 않습니다.
도구가 알 수 없는 배치를 보고하면 준비된 이름을 나열하고 이 관찰에는 demo를 사용하세요.
bigdata-python -B vote_tool.py list
오타가 broker를 초기화한 것은 아닙니다. 인수를 수정하고 읽기 전용 검사를 반복하세요. 표시된 집계를 고치려고 임의의 이벤트를 다시 보내지 마세요.
논리적 식별 정보를 유지하고 내용 충돌 거부하기
이 단계에서는 준비된 Python 함수 안에 비즈니스 규칙을 구현합니다. Terminal의 /home/labex/project에서 계속 작업하세요. 현재 파일을 읽습니다.
cat reducer.py
이 함수는 카탈로그 집계를 초기화한 뒤 입력 값마다 집계를 늘립니다. 데모의 레코드 세 개를 처리하면 어떻게 될지 예상해 보세요. 두 번째로 저장된 demo-1이 표를 하나 더 추가할 것입니다. 올바른 규칙은 이 ID를 한 번만 세면서 서로 다른 ID는 분리해야 합니다.
준비된 규칙 검사기는 함수의 정확한 복사본을 일회용 예제로 호출하며 값의 순서를 바꾸거나 다른 부스 이름도 사용합니다. 실제 Kafka 로그는 바꾸지 않습니다. 현재의 단순한 함수를 실행하세요. 다음 명령은 규칙 메시지와 함께 실패하는 것이 정상입니다.
bigdata-python -B check_rules.py
셸은 바로 이전 명령의 종료 상태를 $?에 저장합니다. 0은 성공, 0이 아닌 값은 실패입니다. 다른 명령이 바꾸기 전에 즉시 표시하세요.
printf '%s\n' "$?"
summarize(values, booths)는 유효한 입력에 votes와 counts가 들어 있는 딕셔너리를 반환하고, 무효한 입력에는 일반 예외를 발생시켜야 합니다. 모듈은 함수 호출이 가능한 상태를 유지해야 하며 그 안에서 Python 인터프리터를 종료하지 마세요. 검사기는 인터프리터의 종료 상태만이 아니라 실제 반환값을 받아야 합니다.
nano에서 기존 함수 편집하기
Terminal의 텍스트 편집기로 편집 가능한 파일을 여세요.
nano reducer.py
위쪽 표시줄에 파일 이름이 표시됩니다. 화살표 키로 이동하고 Backspace/Delete로 문자를 고칩니다. Python은 들여쓰기로 문장을 묶습니다. 함수 안에서는 공백 네 개, 반복문 안에서는 여덟 개를 사용하고 조건을 중첩할 때마다 네 개를 더합니다. 기존 def summarize(values, booths): 행은 유지하세요. 해당 로직을 교체하면서 의도적으로 레코드를 센다는 기존 주석은 제거합니다.
다음의 짧은 Python 블록은 이 함수 하나 안에서 이어지는 편집입니다. 셸에서 실행할 명령이나 별개의 프로그램이 아닙니다. 검사를 실행하기 전에 편집을 모두 마치세요. 파일 전체를 교체하지 말고 시작 코드의 올바른 부분은 유지합니다.
먼저 기존의 0표 초기화를 유지하고, 그 바로 아래이면서 반복문 앞에 seen을 추가하세요. event_id를 키로 하는 딕셔너리가 승인된 각 비즈니스 투표의 전체 값을 보관합니다.
counts = {booth: 0 for booth in booths}
seen = {}
다음으로 기존의 단 하나인 for value in values: 행을 유지하고 그 안의 원래 증가 문장을 아래 검증으로 교체합니다. 블록은 맥락을 보여 주기 위해 기존 for 행을 포함합니다. 두 번째 반복문을 추가하지 마세요. set(value)는 필드 이름을 모으고, isinstance는 타입을 검사하며, any(...)는 필드 값 중 적어도 하나가 유효하지 않은지 확인합니다. 첫 조건은 비어 있지 않은 문자열 필드가 정확히 세 개이고 허용된 선택값을 가진 딕셔너리를 요구합니다. 부스 집계를 늘리기 전에 카탈로그를 검사하세요. for 행은 공백 네 개, 그 안의 검증은 여덟 개로 들여씁니다.
for value in values:
if (not isinstance(value, dict)
or set(value) != {"event_id", "booth", "choice"}
or any(not isinstance(item, str) or not item for item in value.values())
or value["choice"] != "favourite"):
raise ValueError("Invalid vote schema")
if value["booth"] not in counts:
raise ValueError("Unknown booth: " + value["booth"])
이제 카탈로그 검사 바로 뒤이면서 여전히 반복문 안에 식별 처리를 추가하세요. 이미 아는 ID라면 디코딩한 전체 값을 비교합니다. 내용이 다르면 해당 ID를 명시한 오류를 발생시켜야 합니다. 첫 값이나 마지막 값을 선택하면 결과가 순회 순서에 따라 달라집니다. 동일한 재전송은 continue에 도달하여 증가 없이 다음 입력으로 넘어갑니다. 처음 본 ID는 저장하고 집계합니다.
event_id = value["event_id"]
if event_id in seen:
if seen[event_id] != value:
raise ValueError("Conflicting event_id: " + event_id)
continue
seen[event_id] = dict(value)
counts[value["booth"]] += 1
dict(value)는 승인된 값을 딕셔너리로 보관합니다. JSON 공백과 필드 순서는 비즈니스 식별 기준이 아닙니다. 딕셔너리의 키는 선택한 배치의 파티션 전체에서 전역적이며 부스 이름이나 파티션/오프셋 쌍이 아닙니다. 부스와 선택값이 같아도 ID가 다르면 각각 처음 본 ID의 증가 문장에 도달해야 합니다.
마지막으로 모든 입력 값을 사용하던 기존 return을 교체하세요. 승인된 고유 값들과 카탈로그 전체 집계를 반환합니다. 이 행은 반복문 밖이면서 함수 안에 있으므로 공백 네 개로 들여씁니다.
return {"votes": list(seen.values()), "counts": counts}
저장하고 종료하세요. nano 아래쪽 표시의 ^는 Ctrl을 누르라는 뜻입니다. Ctrl+O, Enter로 reducer.py를 확인하고 Ctrl+X로 셸에 돌아옵니다. 이 키들은 현재 파일을 저장하며 Python을 실행하는 것은 아닙니다. 실제 저장한 내용을 다시 읽으세요.
cat reducer.py
seen이 함수 호출당 한 번 초기화되는지, 집계 전에 스키마/카탈로그를 검사하는지, 처음 본 ID만 집계를 늘리는지, return이 반복문 밖에 있는지 확인하세요. 이어서 바로 완성된 함수를 시험합니다.
bigdata-python -B check_rules.py
식별 정보, 서로 다른 ID, 충돌, 스키마, 0표 부스, 순서를 바꾼 입력에 대해 Rules pass가 보여야 합니다. 부스만 키로 하는 딕셔너리는 서로 다른 투표 이벤트를 합치고, 파티션/오프셋을 키로 하는 딕셔너리는 모든 재전송을 남깁니다. 여러분의 딕셔너리는 비즈니스 식별 기준을 나타냅니다.
들여쓰기, 문법, 이름 오류가 있으면 보고된 파일 이름과 행을 읽고 nano reducer.py를 다시 실행한 뒤 화살표 키로 해당 부분을 고치세요. Ctrl+O, Enter로 저장하고 Ctrl+X로 종료한 다음 cat reducer.py로 확인하고 규칙 검사를 다시 실행합니다. 이미 올바른 부분은 유지하세요. 함수 전체를 다시 작성하거나 VM을 초기화할 필요는 없습니다. 경로가 맞지 않으면 다음 읽기 전용 명령으로 찾을 수 있습니다.
pwd
ls reducer.py replay.py
필요하면 cd /home/labex/project로 돌아오세요. 다음 단계에서는 준비된 네이티브 읽기 프로그램으로 실제 Kafka 레코드에 규칙을 적용합니다.
실제 갤러리 로그에서 스냅샷을 내보내고 읽기
이 단계에서는 실제로 고정된 갤러리 배치에 함수를 적용합니다. Terminal의 /home/labex/project에서 계속 작업하세요. main 배치가 선택한 원본과 출력을 나열한 다음 카탈로그를 확인합니다.
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main은 gallery-votes를 가리킵니다. 여러 파티션, 실제로 다른 저장 위치에 있는 반복 ID, 나머지 내용은 같지만 ID가 다른 이벤트, 빈 파티션이 있습니다. replay.py는 준비된 네이티브 읽기 프로그램이며, vote_support.py는 인수 해석과 스냅샷 기록을 제공합니다. 읽기 프로그램은 각 실제 레코드 전체 정보의 값을 summarize 함수에 전달합니다. 로컬 투표 입력 파일에서 값을 가져오지 않습니다.
아래의 공개 진입점을 실행하세요. 먼저 현재 식별 함수를 일회용 예제로 검사한 다음, 현재 프로그램과 보조 자식 프로세스를 실행하고, 그 프로세스 트리의 네이티브 읽기 증거를 보관하며, 성공한 결과를 표시된 출력 경로에 저장합니다.
bigdata-python -B run_replay.py main
records=8, logical_votes=5 결과가 기록되어야 합니다. 진입점은 같은 클라이언트 이름을 쓰는 무관한 컨슈머를 신뢰하는 대신, 이 프로그램 트리가 실제로 broker 데이터를 받았음을 기록합니다. 내부 계산이 실제로 그 입력에 근거했음이나 관리자의 위조에 대한 보호를 증명하지는 않습니다. 최종 비즈니스 결과는 별도로 실제 전체 원본과 비교합니다.
성공한 각 실행은 그 호출 중에 결과를 실제로 기록해야 합니다. 파일 데이터를 직접 쓰거나, 임시 파일에 쓴 뒤 결과 파일로 이름을 바꿀 수 있습니다. 보조 자식 프로세스가 써도 됩니다. 같은 바이트를 다시 기록하는 것은 유효하지만, Kafka를 읽고 오래된 파일을 유지하기만 해서는 새 결과 기록이 아닙니다.
실제 결과 파일을 읽으세요. 이 읽기 도구는 누락된 결과를 조용히 가져오거나 채우거나 복구하지 않습니다.
bigdata-python -B read_snapshot.py output/main-snapshot.json
다섯 개의 논리적 투표 ID와 다음 부스별 집계가 보여야 합니다.
Booth Votes
fern 2
oak 3
reed 0
스냅샷에는 물리 레코드 여덟 개의 전체 정보도 보관되어 있어, 원본 레코드와 논리적인 다섯 표를 구분할 수 있습니다. reed는 0표로 남습니다. 준비된 기록 방식을 사용하면 결과 기록 이전의 루프나 검증 오류가 이전 스냅샷을 사용할 수 있게 남겨 둡니다. 진입점이 다른 쓰기 방식을 자동으로 롤백하는 것은 아닙니다. 프로그램을 편집한 뒤에는 완료 검사 전에 공개 진입점을 다시 실행하세요.
원래 표시 애플리케이션은 컨슈머 그룹의 진행 상황을 저장한 채 일시 중지되어 있습니다. 그 그룹의 설명을 볼 수는 있지만, 가입하거나 초기화/삭제하거나 그 그룹에 오프셋을 커밋해서는 안 됩니다. 원본 토픽을 편집/삭제하거나 파티션을 추가하지 마세요. 자신의 파생 출력만 재구성하세요. 검사는 원래 토픽 식별 정보, 전체 파티션/레플리카 토폴로지, 실제 레코드, 표시 애플리케이션의 저장된 진행 상황을 확인합니다. 임의의 broker 설정 보존을 보장하지는 않습니다.
이제 단순한 전달 내역이 아니라 유용한 부스 스냅샷을 얻었습니다. 다음 실습에서는 빈 폴링 한 번이 배치 완료를 뜻하지 않는 이유를 포함해, 완전하고 독립적인 네이티브 읽기 프로그램을 설명하고 만듭니다.
요약
실제로 저장된 재전송을 관찰하고, 전역적인 비즈니스 식별 정보를 Kafka 레코드 위치와 구분했으며, 같은 ID의 충돌하는 내용을 거부했습니다. 실제 갤러리 스냅샷은 전체 물리 원본을 보관하면서 논리적인 표와 0표를 포함한 모든 부스의 집계를 보여 줍니다. 원본과 일시 중지된 표시 애플리케이션의 그룹도 그대로 사용할 수 있습니다.
