Kafka 데이터 예시

이 페이지에서는 MongoDB 를 공부하기 위해 필요한 용어들 및 데이터 예시를 정리한다.

또한 이 페이지는 perplexity 를 통해 필자가 어떤 prompt 를 작성하여 공부하는지를 공유한다. ([] 로 표기)

그리고 확인 하는 내용에 개인적인 첨언이 필요할시에는 ※ 를 사용한다.

 

[Kafka 에 기록되는 데이터 예시를 보여줘]

 

Kafka에 기록되는 데이터 예시

 

Kafka에 기록되는 데이터(메시지)는 일반적으로 **Key(키)**와 **Value(값)**로 구성되며, 필요에 따라 **Timestamp(타임스탬프)**와 같은 추가 정보가 포함될 수 있다. 값(Value)은 문자열, JSON, 바이너리 등 다양한 형식이 될 수 있다.

 

1. 단순 문자열 메시지
json
{
  "key": "user123",
  "value": "로그인 성공",
  "timestamp": "2025-05-15T02:10:00Z"
}

2. JSON 형식의 메시지 (예: 주문 데이터)
json
{
  "key": "order_10001",
  "value": {
    "orderId": 10001,
    "userId": "user123",
    "product": "노트북",
    "quantity": 1,
    "price": 1200000,
    "orderDate": "2025-05-15T02:10:00Z"
  }
}

3. 로그 데이터 메시지
json
{
  "key": "webserver1",
  "value": {
    "level": "INFO",
    "message": "API 요청 성공",
    "timestamp": "2025-05-15T02:10:00Z",
    "path": "/api/v1/login"
  }
}

4. IoT 센서 데이터 메시지
json
{
  "key": "sensor_001",
  "value": {
    "temperature": 23.5,
    "humidity": 60,
    "timestamp": "2025-05-15T02:10:00Z"
  }
}

 

이처럼 Kafka에 기록되는 데이터는 키와 값의 쌍으로, 값 부분에 다양한 형태의 데이터를 담을 수 있다. 실제로는 JSON, Avro, Protobuf 등 다양한 직렬화 포맷을 사용할 수 있다.

 

Kafka는 이러한 메시지들을 토픽 단위로 저장하며, 각 메시지는 파티션에 분산되어 저장된다.

실제 비즈니스 상황에서는 주문, 결제, 로그, 알림, 센서 데이터 등 다양한 형태의 메시지가 Kafka에 기록된다.

 

[Kafka 의 JSON 과 MongoDB JSON 형식의 차이점]

 

Kafka에서 사용하는 JSON과 MongoDB에서 사용하는 JSON 형식에는 몇 가지 중요한 차이점이 존재한다.

 

1. 데이터 표현 및 타입 지원

 

• Kafka에서 주로 사용하는 JSON은 표준 JSON 형식으로, 문자열, 숫자, 배열, 객체, 불리언, null 등 기본적인 타입만 지원한다. 표준 JSON은 날짜, 바이너리, ObjectId 등 MongoDB 특유의 타입을 직접적으로 표현하지 못한다.
  
  
• MongoDB는 내부적으로 BSON(Binary JSON) 포맷을 사용하며, BSON은 표준 JSON보다 더 다양한 타입(예: 날짜, 바이너리, ObjectId, Decimal128 등)을 지원한다. 이를 JSON으로 변환할 때 MongoDB는 Extended JSON(EJSON)이라는 확장 표기법을 사용해 BSON 타입을 JSON으로 안전하게 변환한다. 예를 들어, ObjectId는 {"$oid": "..."} 형태, 날짜는 {"$date": ...} 형태로 표현된다.
  

2. 직렬화 방식

 

• Kafka에 기록되는 JSON 데이터는 일반적으로 UTF-8 문자열로 직렬화된다. 즉, 메시지는 텍스트 기반의 JSON 문자열로 저장되고 전송된다.
  
  
• MongoDB는 데이터를 저장하거나 네트워크로 전송할 때 BSON이라는 바이너리 포맷을 사용한다. 애플리케이션에서 데이터를 읽거나 쓸 때는 BSON과 JSON 간의 변환이 필요하다.
  
  

3. 데이터 타입 보존

• Kafka의 표준 JSON 메시지는 타입 정보를 명확히 보존하지 못한다. 예를 들어, 숫자 타입의 구분(정수, 부동소수점), 날짜, 바이너리 등은 모두 문자열이나 숫자로만 표현된다.
  
  
• MongoDB에서 Kafka Connector를 사용할 경우, JSON 포맷터 설정에 따라 BSON 타입 정보를 보존하는 Extended JSON, 기본 JSON, 단순화된 JSON 등 다양한 변환 방식을 지원한다. Extended JSON을 사용하면 BSON 타입이 JSON 내에서 명확히 표현되어 데이터 타입의 손실 없이 저장 및 전송이 가능하다.

 

※ Kafka 의 문자열 숫자 표기는 이후 다른 게시글에 언급할 AWS DMS (Database Migration Service) 에서 주의 해야 할 특징 중하나다.

 

[Kafka 는 RDBMS 로부터 데이터를 받을 때 어떻게 보관하도록 설정할까?]

 

Kafka는 RDBMS로부터 데이터를 받을 때 Kafka Connect의 JDBC Source Connector를 주로 사용해 데이터를 수집하고 보관하도록 설정한다.
JDBC Source Connector는 RDBMS에 주기적으로 연결해 SQL 쿼리를 실행하고, 결과 데이터를 Kafka의 토픽(Topic)에 메시지 형태로 저장한다.

데이터를 보관하는 방식은 다음과 같다.

• 연결 및 추출 방식
  
  JDBC Source Connector를 통해 RDBMS(MySQL, PostgreSQL, Oracle 등)에 연결하고, 지정한 테이블이나 쿼리 결과를 주기적으로 조회한다.
  추출 방식에는 전체 데이터 추출(Bulk), 사용자 정의 쿼리(Custom Query), 그리고 증분 추출(Incremental)이 있다. 증분 추출은 보통 증가하는 PK 컬럼이나 타임스탬프 컬럼을 기준으로 신규 또는 변경된 데이터만 추출한다.
  
  
• Kafka 토픽에 저장
  
  추출된 데이터는 Kafka의 토픽에 메시지로 저장된다. 이때 각 메시지는 일반적으로 JSON, Avro, String 등 지정한 포맷으로 직렬화된다.
  메시지의 Key와 Value에는 테이블의 PK, 데이터 전체 Row, 타임스탬프 등 원하는 정보를 담을 수 있다.
  
  
• 설정 예시
  
  다음과 같이 Connector를 설정해 사용할 수 있다.
  

json
{
  "name": "my-source-connect",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://localhost:3306/source_db",
    "connection.user": "root",
    "connection.password": "1111",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "table.whitelist": "users",
    "topic.prefix": "my_topic_",
    "tasks.max": "1"
  }
}

 

• 위 설정은 MySQL의 users 테이블에서 id 컬럼을 기준으로 신규 데이터만 추출해 my_topic_users라는 Kafka 토픽에 저장하도록 한다.
  
  
• 데이터 포맷
  Kafka에 저장되는 데이터는 기본적으로 JSON, Avro, String 등으로 직렬화된다.
  실제 메시지 예시는 다음과 같다.
  
  

json
{
  "id": 101,
  "name": "홍길동",
  "email": "hong@example.com",
  "created_at": "2025-05-15T02:36:00Z"
}

 

요약하면, Kafka는 RDBMS에서 데이터를 받을 때 Kafka Connect의 JDBC Source Connector를 사용해 주기적으로 데이터를 추출하고, 지정한 Kafka 토픽에 메시지 형태로 저장하도록 설정한다. 추출 방식, 데이터 포맷, 토픽명 등은 설정 파일에서 자유롭게 지정할 수 있다.