👨‍👧‍👦10. 데이터 동기화를 위한 Apache Kafka활용하기 (1)

🍀 kafka 의 개요

• Scalar 언어로 된 오픈 소스 메세지 브로커 프로젝트이다
• 실시간 데이터 피드를 관리하기 위해 통일된 높은 처리량과 낮은 지연 시간을 지닌 플랫폼을 제공하고 있다.

♪ kafka 를 사용하는 이유

• 위 사진의 방식의 단점
  • End - to - End 연결 방식의 아키텍처이다
  • 데이터 연동의 복잡성이 증가한다 (HW, 운영체제, 장애 등) 
  • 서로 다른 DB를 사용하면 다른 종류의 DB에 데이터를 전달하는 것은 쉽지 않다.
  • 때문에 확장이 어렵다는 단점이 존재한다.
• 때문에 모든 시스템으로 데이터를 실시간으로 전송하여 처리하는게 필요했고 또 데이터가 많아지더라도 확장이 용이한 시스템이 필요졌고 그로 인해 Kafka라는 시스템이 도입되었다

• Kafka의 등장으로 각각의 DB는 더이상 어떠한 시스템으로 전달되는지 상관하지 않고 데이터만 전달하면 되는 방식으로 전환되었다. 
• 또 각각의 서비스는 더이상 DB의 연동을 신경쓰지 않고 Kafka에게 데이터를 받아오기만 하게 되면서 단일 포맷을 유지할 수 있게 된다.

 

---

 

🍀 kafka Broker

• kafka의 서버를 뜻한다.
• 3대 이상의 broker cluster로 구성되어 있다.
  • 혹여 한대의 브로커가 문제가 발생하였을 경우 대신할 브로커를 대비한다.
• Zookeeper와 연동된다
  • 역할 : 메타데이터 (Broker ID, Controller ID 등) 저장된다.
  • Controller 정보도 저장된다
• N개의 Broker 중 1대는 Controller 기능을 수행한다.
  • Controller(리더) 역할
    • 각 Broker 에게 담당 파티션 할당을 수행한다
    • Broker 정상 동작 모니터링을 관리 할 수 있다.

ZooKeeper란?
Apache 제간에서 개발되었고 분산 응용 프로그램을 위한 오픈 소스 분산 코디네이션 서비스이다.
여러 대의 컴퓨터 간의 안전하고 정확한 데이터 동기화를 제공하게 된다. 
1. 분산 구성 관리 : 여러 노드에서 실행되는 어플리케이션의 구성을 동기화하고 관리한다.
2. 분산 락 제공 : 여러 클라이언트가 공유 자원에 대한 접근을 조율하기 위해 분산 락을 사용할 수 있다. 이로 인해 임계 영역에 대한 동시 액세스 문제를 해결한다
3. 분산 메세지 큐 : 이벤트 및 메세지 큐 서비스를 제공하여 분산 어플리케이션 간에 데이터를 안전하게 전달한다.
4. 리더 선출 : 리더 노드를 정하고 장애 발생 시 새로운 리더를 빠르게 선출 한다.

 

---

 

🍀 kafka Client ( Producer & Consumer )

두가지의 시나리오를 구성하여 Kafka 를 사용해보자

 

♪ Kafka의 서버 기동

• Zookeeper와 Kafka를 각각 구동해주면 된다.
• Kafka 가 데이터를 보내면 해당 데이터는 Topic에 저장된다.
  • producer가 Topic에게 전달한 내용물이 있다면 해당 Topic을 등록한 Consumer들에게 일괄적으로 메세지가 전달된다.
  • --topic {토픽이름}
  • --bootstrap -server {카프카 서버 주소} : 해당 주소에 토픽을 생성하겠다
  • /--partition 1: 멀티 클러스터링 구조를 생성했을때 토픽에 전달된 메세지를 몇 군데 나눠서 저장할지 정하는 옵션

 

---

 

🍀 시나리오 1 ( Producer & Consumer의 메세지 전달 )

Kafka에 메세지를 전달하고 Kafka가 메세지를 저장하고 있다가 다른 Consumer에게 메세지 전달

• 위에 과정에서 생성된 Topic을 통해 시나리오를 시작한다
• kafka-console-producer : 보내는 자
  • --broker-list : 어디에 메세지를 전달할 것인가
  • --topic : 생성하고 하는 토픽의 이름
• Kafka-console-consumer : 받는자
  
  • --bootstrap-server : 받고자 하는 주소의 이름
  • --topic : 연결하고자 하는 토픽의 이름
  • --from-beginning : 메세지를 처음부터 받아오겠다는 것
• 해당 사진은 두개의 cmd 창을 열어 진행한 것이다.

 

---

 

🍀 시나리오 2 ( Connect를 이용한 데이터 전달 )

♪ Kafka 

특별한 프로그래밍을 사용하지 않고 Configuration만 가지고 데이터를 특정한 곳에서 받아와 다른쪽으로 이동시켜주는 기능이다.

• Kafka Connect를 통해 Data를 Import / Export가 가능하다
• 코드 없이 Configuration으로 데이터 이동이 가능하다
• Stadalone mode, Distrivution mode 를 지원한다
  • RESTful API 를 통해 지원한다
  • Stream 또는 Batch 형태로 데이터 전송이 가능하다
  • 커스텀 Connector를 통해 다양한 Plugin을 제공한다.
• 파일로부터 데이터를 받고(Import) 파일로 데이터를 전송(Export)하고 DB로부터 데이터를 받아와 파일로 데이터 전송할 수잇다.
• kafka connect source 는 데이터를 가져오고 kafka connect sink는 데이터를 보내는 쪽이라고 생각하면된다.

♪ MariaDB 연동

• MariaDB를 설치해준다
• MySQL Client(cmd)를 사용해서 DB를 설치할때 설정해준 비밀번호를 통해 로그인을 한다.
• 원하는 database 를 생성하고 테이블은 해당 서비스에 mariadb의 의존성을 추가시키고 h2 console을 이용해서 쿼리문을 좀 더 쉽게 작성할 수 있다.
• 그리고 다시 cmd 를 사용해서 확인하면 결과를 확인 할 수있다.

 

• 또한 우리가 Kafka의 데이터 전송 실습을 위한 서비스인 orderservice 에 마리아DB 라이브러리를 추가해준다.

♪ Kafka Connect 와 JDBC Connect 를 이용해 연결하는 설치 방법

설치를 하는 과정은 따로 블로그로 작성해두었다. 워낙 고생을 많이 해서 따로 정리를 했다.

https://latewalk.tistory.com/211

👨‍👧‍👦 Kafka 와 JDBC Connect 설치 및 환경 확인

 

1. 체크

postman을 통해 데이터를 바로 체크하기 전에 http://localhost:8083/connector-plugins 라는 url 을 통해 jdbcSourceConnector라는 class 가 있는지 확인하자

2. 실행

• 이후 모든 작업이 완료되었다면 이렇게 요청한 작업이 정상적으로 진행되는 것을 알 수 있다.
• 또한 같은 요청을 GET을 통해 받아보면 name으로써 지정한 커넥트가 보인다.

• 이 내용을 기반으로 status를 확인하면 상황이 끝난다.

 

♪ 시나리오 2 시작

DB에 있는 자료가 변경사항이 생겼을 경우 Kafka가 DB로부터 변경된 값을 가져오고 그 값을 다른 DB, 스토리지, 서비스에 전달해주는 카프카 커넥트 기능

1. 데이터 넣기

• users 에 데이터를 추가적으로 입력해준다.

 

2. 토픽 확인

• topic 으로 생성된 users가 존재한다. 근데 분명히 앞에 my_topic 을 넣게 만들었는데 왜 안됐지..

 

3. 컨슈머를 통해 확인

• 놀라운것은 이제 mariaDB를 통해 데이터를 추가하면 컨슈머가 알아서 인지하여 바로 확인한다는 것이다.

• 분명히 위에서는 한개였는데 admin을 추가해주고 엔터를 누르니 바로 컨슈머쪽에서도 확인이 되는 것을 볼 수 있다.
• 한줄로 작성되어 알아보기 어려우니 정리해보면 이렇게 볼 수 있다.

 

 

---

 

🍀 Sink Connect

• 이전에 우리가 Source Connect 를 통해 데이터를 Topic에 전달하는 것을 확인 할 수 있었다. 
• 이 다음의 SinkConnector 가 하는 일은 Topic 에 전달된 이 데이터를 사용하는 곳으로 전달하는 것이라고 할 수 있다

• 이렇게 제작이 되었다면 table과 같은 형태의 table이 생성되었다는 뜻이다.
• 그리고 DB를 확인해보면 토픽의 이름과 같은 테이블이 하나 생성 된것을 알수 있다.

• source 에서 제작된 my_topic_users 라는 이름이 테이블로 전환되어 저장된것을 볼 수 있다. 이는 전달된 내용까지도 똑같이 생성되며 users 테이블과 my_topic_users는 같은 데이터를 주고 받는것을 알 수있다.

• 기존의 users 테이블과 연결된 SourceConnect 는 Topic 을 제작하여 해당 토픽에 데이터를 전달한다. 그럼 컨슈머의 역할 하고 있는 아이는 해당 변경사항을 그때그때 전달받는다. 
• 중요한것은 sinkConnect가 제작되고나서이다. SinkConnect는 토픽을 연결하고 해당 토픽의 이름을 기준으로 sink와 연결되어있는 DB에 테이블을 제작하고 데이터를 그대로 전달한다. 물론 컨슈머가 읽은 그 데이터 모두를 말하는 것이다.
• 또한 데이터를 producer로 직접 작성해서 전달을 하더라도 똑같이 컨슈머는 읽을것이고 sink를 통해 my_topic_users의 테이블에 변경이 되는 것을 알 수 있다. 하지만 그냥 users에는 테이블의 변경이 없다. 즉, users의 테이블에 데이터가 들어가게 되면 해당 데이터를 JSON 형식으로 전환하여 전달한다는 것으로 알 수 있는 것이다.