현대오토에버 부트캠프에서 진행한 Connected Car Telemetry 프로젝트를 우수상과 함께 마무리했습니다.
당시 프로젝트는 3주 동안 진행한 Kubernetes 미니 프로젝트였습니다.
짧은 기간 안에 CQRS 패턴과 Kubernetes 기반 가용성 관리에 집중하다 보니, 데이터 엔지니어링 관점에서는 아쉬운 부분이 남았습니다.
가장 아쉬웠던 점은 데이터 유실 방지를 충분히 설계하고 검증하지 못했다는 것이었습니다.
기존 프로젝트에서는 Kafka를 중심으로 MongoDB에는 차량 운행 이력을, Redis에는 차량의 최신 상태를 저장했습니다.
하지만 Producer와 Consumer 장애, Kafka Broker 장애, 중복 처리와 재처리 같은 상황을 충분히 실험하지 못했습니다.
따라서 기존 프로젝트의 아이디어를 가져오되, 이번에는 데이터 엔지니어 관점에서 처음부터 다시 구성해보기로 했습니다.
목표는 단순히 기술을 연결하는 것이 아니라 다음 질문에 답할 수 있는 파이프라인을 만드는 것입니다.
데이터가 언제 생성되고, 어디를 거쳐 저장되며, 장애가 발생했을 때 데이터가 유실되거나 중복되지 않았다는 것을 어떻게 증명할 것인가?
1. 차량 데이터는 어디서 가져올까?
실제 커넥티드카의 대규모 텔레메트리 데이터를 개인 프로젝트에서 구하기는 쉽지 않았습니다.
처음에는 직접 더미 데이터 생성기를 만들려고 했습니다.
하지만 차량의 위치와 속도, 가속도, 도로와 차선 정보까지 임의로 생성하면 현실적인 데이터라고 보기 어려웠습니다.
그러던 중 찾게 된 것이 오픈소스 교통 시뮬레이터인 SUMO였습니다.
SUMO는 도로 네트워크와 차량 경로를 바탕으로 차량의 움직임을 계산합니다.
차량 수와 시뮬레이션 시간 등을 조절할 수 있기 때문에 실시간 데이터 파이프라인의 부하 테스트에도 활용할 수 있습니다.
SUMO의 OSM Web Wizard를 이용해 서울 강남역 주변 도로를 가져왔습니다.
이번 시나리오에서는 차량 데이터에만 집중하기 위해 대중교통, 보행자, 자전거 등은 제외하고 승용차만 생성했습니다.
시뮬레이션을 실행하면 차량이 각자의 출발 시간과 경로에 따라 도로 위를 이동합니다.
시뮬레이션 도중 활성 차량 수가 12대에서 20대 이상으로 늘어났다가 다시 줄어드는 모습도 확인할 수 있었습니다.
이는 차량이 고정된 수로 계속 존재하는 것이 아니라, 새로운 차량이 출발하면 활성 차량 수가 증가하고 목적지에 도착하면 시뮬레이션에서 빠지기 때문입니다.
2. SUMO 데이터는 어떻게 Python으로 가져올까?
SUMO에서 차량 데이터를 HTTP로 받아오는 것은 아닙니다.
OSM Web Wizard를 실행했을 때 나타난 127.0.0.1 웹페이지는 지도와 시나리오를 설정하기 위한 화면입니다.
실제 차량 데이터는 SUMO의 TraCI(Traffic Control Interface) 프로토콜을 통해 가져옵니다.
현재 데이터 흐름은 다음과 같습니다.
SUMO 시뮬레이션 엔진
→ TraCI TCP 프로토콜
→ Python traci 라이브러리
→ Python DictionaryPython 코드에서 시뮬레이션을 1초 진행한 후 현재 활성화된 모든 차량을 조회했습니다.
traci.simulationStep()
vehicle_ids = traci.vehicle.getIDList()이후 TraCI에서 제공하는 메서드로 필요한 정보만 선택해 가져왔습니다.
speed = traci.vehicle.getSpeed(vehicle_id)
acceleration = traci.vehicle.getAcceleration(vehicle_id)
position = traci.vehicle.getPosition(vehicle_id)
road_id = traci.vehicle.getRoadID(vehicle_id)
lane_id = traci.vehicle.getLaneID(vehicle_id)SUMO가 완성된 JSON을 제공하는 것은 아닙니다.
SUMO가 관리하는 차량 상태 중 필요한 값을 TraCI 메서드로 가져온 뒤, 프로젝트에서 사용할 형태로 직접 가공해야 합니다.
차량 한 대의 데이터는 다음과 같은 Python Dictionary로 만들었습니다.
event = {
"vehicle_id": vehicle_id,
"simulation_time": simulation_time,
"speed_mps": speed,
"acceleration_mps2": acceleration,
"latitude": latitude,
"longitude": longitude,
"angle": angle,
"road_id": road_id,
"lane_id": lane_id,
}현재 수집하는 필드는 다음과 같습니다.
| 필드 | 설명 | 단위 |
|---|---|---|
vehicle_id | 차량 식별자 | 문자열 |
simulation_time | 시뮬레이션 경과 시간 | 초 |
speed_mps | 차량 속도 | m/s |
acceleration_mps2 | 차량 가속도 | m/s² |
latitude | 위도 | 실수 |
longitude | 경도 | 실수 |
angle | 차량 진행 방향 | degree |
road_id | SUMO 도로 식별자 | 문자열 |
lane_id | SUMO 차선 식별자 | 문자열 |
터미널에서 다음처럼 작은따옴표로 보이는 데이터는 아직 JSON이 아니라 Python Dictionary입니다.
{
'vehicle_id': 'veh10',
'simulation_time': 68.0,
'speed_mps': 13.9404
}Kafka로 보내기 전 json.dumps()를 이용해 JSON으로 직렬화했습니다.
message_value = json.dumps(
event,
ensure_ascii=False,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")직렬화된 Kafka 메시지는 다음과 같은 형태가 됩니다.
{
"vehicle_id": "veh10",
"simulation_time": 68.0,
"speed_mps": 13.9404,
"acceleration_mps2": -0.7496,
"latitude": 37.49818,
"longitude": 127.03447,
"angle": 249.53,
"road_id": "117823785#0",
"lane_id": "117823785#0_0"
}3. SUMO Source와 Kafka Producer 분리
데이터 수집과 Kafka 전송의 책임을 분리하기 위해 다음과 같이 파일을 나눴습니다.
src/
├─ sumo_source.py
└─ kafka_producer.pysumo_source.py는 다음 역할을 담당합니다.
- SUMO 실행
- TraCI 연결
- 1초 단위 시뮬레이션 진행
- 현재 활성 차량 조회
- 차량 상태를 Dictionary로 가공
- Kafka Producer 호출
kafka_producer.py는 다음 역할을 담당합니다.
- Dictionary를 JSON으로 직렬화
vehicle_id를 Kafka Message Key로 설정- Kafka 토픽으로 비동기 전송
- 전송 성공 및 실패 callback 처리
- 종료 시 대기 중인 메시지
flush
구조로 표현하면 다음과 같습니다.
SUMO
→ TraCI
→ sumo_source.py
→ Python Dictionary
→ kafka_producer.py
→ JSON
→ KafkaProducer에서는 vehicle_id를 Message Key로 사용했습니다.
self.producer.produce(
topic=self.topic,
key=event["vehicle_id"].encode("utf-8"),
value=message_value,
on_delivery=self._delivery_report,
)같은 vehicle_id는 동일한 파티션으로 전달되기 때문에, 파티션 수가 변경되지 않는다면 같은 차량의 이벤트 순서를 파티션 내부에서 유지할 수 있습니다.
4. 단일 Kafka Broker로 전송 확인
우선 Kafka의 기본 흐름을 확인하기 위해 단일 Broker를 실행했습니다.
Kafka 4.3.1 공식 Docker 이미지를 사용했으며 KRaft 모드로 실행했기 때문에 ZooKeeper는 별도로 구성하지 않았습니다.
Windows에서 9092 포트가 시스템 제외 범위에 포함되어 있어 외부 접속 포트로 29092를 사용했습니다.
생성한 토픽은 다음과 같습니다.
Topic: vehicle-telemetry-v1
Partitions: 3
Replication Factor: 1Producer 설정은 다음과 같습니다.
self.producer = Producer(
{
"bootstrap.servers": "localhost:29092",
"client.id": "sumo-telemetry-producer",
"acks": "all",
"enable.idempotence": True,
"max.in.flight.requests.per.connection": 5,
}
)acks=all
현재 정상적으로 동기화된 모든 Replica가 메시지를 기록했다는 응답을 보내야 전송을 성공으로 처리합니다.
하지만 이 시점에는 Broker와 Replica가 하나뿐이었기 때문에, 실제로는 Broker 한 대의 응답만 확인하는 구조였습니다.
enable.idempotence=True
네트워크 문제로 Kafka Producer가 동일한 전송 요청을 재시도했을 때 Kafka 내부에서 중복으로 기록되는 것을 방지합니다.
단, 애플리케이션 재시작 후 동일한 데이터를 직접 다시 보내는 것까지 제거해주는 것은 아닙니다.
이를 위해서는 별도의 event_id가 필요합니다.
max.in.flight.requests.per.connection=5
Kafka의 응답을 받지 않은 상태에서 하나의 Broker 연결에 최대 몇 개의 요청을 동시에 보낼 수 있는지 설정합니다.
멱등성을 활성화한 상태에서는 이 값을 5 이하로 설정해 재시도 과정에서도 파티션 내부 순서를 유지할 수 있습니다.
Console Consumer를 실행한 결과 SUMO에서 생성된 차량 데이터가 JSON 형태로 Kafka에 들어오는 것을 확인했습니다.
시뮬레이션을 1초 단위로 진행했기 때문에 활성 차량이 20대라면 해당 시점에는 초당 약 20개의 이벤트가 생성됩니다.
5. 단일 Broker의 한계
여기까지 데이터 전송에는 성공했지만, 이 구조는 아직 고가용성 구조가 아닙니다.
Broker 1대
Partition 3개
Replication Factor 1파티션은 3개지만 모두 동일한 Broker 안에 존재합니다.
Broker 0
├─ Partition 0
├─ Partition 1
└─ Partition 2Broker 프로세스가 잠시 중단됐다가 동일한 디스크로 복구되면 데이터를 다시 사용할 수 있습니다.
하지만 Broker의 디스크나 컨테이너 데이터가 손상되면 다른 복제본이 없기 때문에 복구할 수 없습니다.
Broker가 중단된 동안에는 Producer와 Consumer의 읽기·쓰기도 불가능합니다.
따라서 acks=all만 설정한다고 데이터 유실이 방지되는 것은 아닙니다.
6. Kafka 3 Broker 클러스터 구성
단일 Broker의 한계를 개선하기 위해 Docker Compose를 사용해 Kafka Broker를 3개로 확장했습니다.
kafka-1 → localhost:29092
kafka-2 → localhost:39092
kafka-3 → localhost:49092각 노드는 KRaft의 Broker와 Controller 역할을 함께 수행하도록 구성했습니다.
로컬 학습 환경이기 때문에 세 노드가 Broker와 Controller 역할을 함께 담당하지만, 더 큰 운영 환경에서는 Controller 노드를 분리하는 구성도 고려할 수 있습니다.
목표 토픽 설정은 다음과 같습니다.
Partitions: 3
Replication Factor: 3
min.insync.replicas: 2예상되는 파티션 배치는 다음과 같습니다.
| 파티션 | Broker 1 | Broker 2 | Broker 3 |
|---|---|---|---|
| Partition 0 | Leader | Replica | Replica |
| Partition 1 | Replica | Leader | Replica |
| Partition 2 | Replica | Replica | Leader |
각 파티션은 하나의 Leader와 두 개의 Follower Replica를 가지게 됩니다.
Producer의 쓰기 요청은 Leader가 처리하고, Follower들이 Leader의 데이터를 복제합니다.
Leader Broker가 중단되면 동기화된 Replica 중 하나가 새로운 Leader가 됩니다.
min.insync.replicas=2를 적용하면 정상적으로 동기화된 Replica가 2개 미만일 때 Kafka는 쓰기 요청을 성공으로 처리하지 않습니다.
Replication Factor 3
+ min.insync.replicas 2
+ Producer acks=all이를 통해 Kafka가 안전하게 복제할 수 없는 상황에서 성공 응답을 반환하는 것을 방지할 수 있습니다.
7. Broker별 Docker Volume 구성
Kafka 컨테이너가 재생성되더라도 데이터를 유지할 수 있도록 Broker마다 Docker Named Volume을 연결했습니다.
volumes:
kafka-1-data:
kafka-2-data:
kafka-3-data:각 Broker는 자신에게 할당된 볼륨을 Kafka 로그 저장 경로에 연결합니다.
volumes:
- kafka-1-data:/tmp/kraft-combined-logs구조는 다음과 같습니다.
Kafka Broker 1
→ /tmp/kraft-combined-logs
→ kafka-1-data
Kafka Broker 2
→ /tmp/kraft-combined-logs
→ kafka-2-data
Kafka Broker 3
→ /tmp/kraft-combined-logs
→ kafka-3-data컨테이너가 삭제되거나 재생성되어도 Docker Volume을 다시 연결하면 기존 데이터를 사용할 수 있습니다.
단, 다음 명령은 볼륨까지 삭제하므로 주의해야 합니다.
docker compose down -v8. Docker Volume 권한 문제 해결
처음 3 Broker 클러스터를 실행했을 때 세 컨테이너가 모두 종료되는 문제가 발생했습니다.
로그를 확인한 결과 다음 오류가 발생하고 있었습니다.
uid=1000(appuser)
java.nio.file.AccessDeniedException:
/tmp/kraft-combined-logs/bootstrap.checkpoint.tmpApache Kafka 공식 이미지는 보안을 위해 Kafka를 appuser(uid=1000)로 실행합니다.
하지만 Docker가 생성한 볼륨 디렉터리는 root 소유였기 때문에 Kafka 프로세스가 파일을 생성할 수 없었습니다.
Docker Volume 소유자: root(uid=0)
Kafka 실행 사용자: appuser(uid=1000)Kafka 자체를 root로 실행하면 간단하게 해결할 수 있지만, 서버 프로세스를 높은 권한으로 실행하는 것은 보안상 바람직하지 않다고 판단했습니다.
대신 볼륨의 권한만 변경하는 초기화 컨테이너를 추가했습니다.
kafka-volume-init:
image: busybox:1.36
user: "0:0"
command:
- sh
- -c
- chown -R 1000:1000 /kafka-1-data /kafka-2-data /kafka-3-data각 Kafka Broker는 초기화 컨테이너가 정상적으로 종료된 후 실행되도록 설정했습니다.
depends_on:
kafka-volume-init:
condition: service_completed_successfully실행 순서는 다음과 같습니다.
Docker Volume 생성
→ 권한 초기화 컨테이너 실행
→ 볼륨 소유자를 1000:1000으로 변경
→ 초기화 컨테이너 종료
→ Kafka Broker 3개 실행최종적으로 다음 상태를 확인했습니다.
kafka-volume-init Exited (0)
kafka-1 Up
kafka-2 Up
kafka-3 Upkafka-volume-init의 Exited (0)은 오류가 아니라 권한 변경 작업을 성공적으로 마치고 종료됐다는 의미입니다.
이 과정에서 Docker Volume, 컨테이너 실행 사용자, Linux UID/GID가 어떻게 연결되는지 다시 이해할 수 있었습니다.
9. 오늘 구성한 전체 흐름
오늘까지 구성한 데이터 흐름은 다음과 같습니다.
SUMO 시뮬레이션
→ TraCI를 이용한 차량 상태 수집
→ Python Dictionary 가공
→ JSON 직렬화
→ Kafka Producer
→ Kafka 3 Broker 클러스터
→ Broker별 영구 Volume아직 “데이터 유실을 완전히 방지했다”고 말할 수 있는 단계는 아닙니다.
오늘은 Kafka에 데이터를 안전하게 저장하기 위한 기반을 구성했습니다.
다음 단계에서는 실제로 장애를 발생시킨 뒤 데이터가 어떻게 처리되는지 검증할 예정입니다.
다음 단계
다음 글에서는 아래 작업을 진행할 예정입니다.
- 3 Broker 클러스터에
vehicle-telemetry-v1토픽 생성 Replication Factor 3과min.insync.replicas=2확인- Producer의 Bootstrap Server를 세 Broker로 변경
event_id,run_id,sequence_no,event_time추가- 생성 이벤트 수와 Kafka 저장 이벤트 수 비교
- Producer 실행 중 Broker 한 대 강제 중단
- Leader 변경 및 전송 지속 여부 확인
- Broker 복구 후 ISR 복원 확인
단순히 Kafka를 사용했다는 것에서 끝내지 않고, 장애를 직접 발생시키고 데이터 유실과 중복 여부를 수치로 검증하는 것이 이번 재구성 프로젝트의 목표입니다.
