전체 글

Flink Job을 사용한 Kafka 메시지 스트리밍데이터 수집에서 파이썬으로 데이터를 생성하고 kafka에 메시지를 publish 했다. 이제 flink로 kafka에서 메시지를 소비하고 변환 및 집계 연산을 수행해야 한다.일단 kafka에서 파이썬으로 생성한 데이터를 그대로 잘 가져오는지 확인해본다. 환경 구축에서 archetype을 flink-quickstart-java로 지정했다면 프로젝트 파일에 아래처럼 스켈레톤 코드가 작성되어 있다.Flink 공식 문서에 kafka connector를 사용해 데이터를 읽는 방법이 자세하게 쓰여있다. At-least-once, At-most-once, Exactly-once 3가지의 이벤트 처리 보장 중 가장 신뢰할 수 있는 exactly-once 방법으로 ka..

데이터 구조Json 형식으로 kafka에 메시지를 publish한다.keyvaluedeliveryIduuiddeliveryDate배달 주문 날짜userId배달 주문 고객 idfoodCategory음식 카테고리foodPrice음식 총 가격paymentMethod결제 방식deliveryDistance배달 거리deliveryDestination배달 목적지destinationLat배달 목적지의 위도destinationLon배달 목적지의 경도deliveryCharge배달 요금 배달 목적지의 위도, 경도를 데이터에 포함하는 이유는 elasticsearch의 지도를 활용하기 위해서다. 배달 목적지에 따른 주문 빈도수를 지도로 한번에 나타내면 좋을 것 같아 위도, 경도를 저장하기로 했다.데이터 생성faker = Fak..

Docker composeFlink를 제외한 서비스들은 전부 docker 컨테이너로 수행된다. 로컬 환경이고, 작은 규모의 데이터를 처리하기 때문에 flink는 standalone 모드로 실행할 예정이나 대규모 데이터 처리, 고가용성이 요구된다면 적합하지 않은 방법이다. docker compose version은 3.8로 지정했다.Zookeeper & Kafka & Control Centerzookeeper: image: confluentinc/cp-zookeeper:7.4.0 hostname: zookeeper container_name: zookeeper ports: - "2181:2181" environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEE..

시스템 구조 Kafka, flink를 사용해 실시간으로 발생하는 배달 주문 데이터를 처리하고 저장한다. 데이터는 postgres, elasticsearch에 저장되며 저장된 데이터를 kibana에서 시각화한다. 환경 구축 - Docker compose 사용해 컨테이너 구성 - Zookeeper, Kafka broker, Control center, Postgres, Elasticsearch, Kibana 컨테이너 실행 및 관리 데이터 수집 - Python 코드에서 배달 주문 데이터 생성 - Kafka 사용해 메시지 publish - Confluent control center에서 kafka 모니터링 데이터 처리 - Flink job 사용해 배달 주문 내역이 저장된 토픽에서 메시지 소비, 실시간 처리 - ..

Broker, Replication, In-Sync Replica Broker, Replication, ISR(In-Sync Replica)은 kafka 운영에 있어 아주 중요한 역할을 한다. kafka 아키텍처의 핵심인 replication(복제)은 클러스터에서 서버에 장애가 생겼을 때 kafka의 가용성을 보장하는 가장 좋은 방법이다. Kafka Broker Kafka broker란 kafka가 설치되어 있는 서버 단위이다. 보통 3개 이상의 broker를 구성해 사용하는 것을 권장한다. Kafka Replication replication은 partition의 복제이다. replication : 1 → partition 1개만 존재 replication : 2 → 원본 partition 1개 + 복..

아파치 카프카 입문 강의를 보고 정리합니다. Topic 데이터가 들어가는 공간을 topic이라고 부른다. Kafka topic은 일반적인 AMQP(Advanced Message Queing Protocol)와는 다르게 동작한다. kafka에서는 topic을 여러개 생성할 수 있으며, DB의 테이블이나 파일 시스템의 폴더와 유사한 성질을 가진다. topic은 이름을 가질 수 있으므로 어떤 데이터를 담는지, 목적에 따라 명확하게 명시하면 추후 쉽게 유지보수가 가능하다. Topic 내부, Partition 하나의 topic은 여러개의 partition으로 구성될 수 있다. partition 번호는 0번부터 시작하며, queue와 같이 끝에서부터 데이터가 차곡차곡 쌓인다. 한 topic에 consumer가 붙게..

아파치 카프카 입문 강의를 보고 정리합니다. Before Kafka 데이터를 전송하는 Source Application과 데이터를 받는 Target Application이 존재하며, 초기에는 단방향 통신을 사용했다. 시간이 흐름에 따라 Source와 Target application이 많아지면서 데이터 전송 라인 또한 복잡해졌다. 데이터 전송 라인이 복잡해지면 배포, 장애에 대응하기 어려워진다. 또한 데이터를 전송할 때 프로토콜, 포맷의 파현화가 심해지며, 데이터 포맷 변경 사항이 있을 때 유지보수도 어려워진다. 이러한 문제들을 해결하기 위해 LinkedIn에서 Apache Kafka를 개발했고 현재는 오픈소스로 제공되고 있다. Kafka 주변 생태계 Kafka는 source와 target applica..

Airflow 마스터 클래스 강의 수강, 개인 공부 후 기록합니다. Macro 변수 필요성 dag이 실행되는 시점에 값이 들어가야 할 때 template 변수를 사용했다. DB 테이블이 존재하고 daily ETL 처리를 위한 조회 쿼리가 매일 0시에 실행된다고 가정할 때 data_interval_start, data_interval_end template 변수를 사용하면 매일 날짜를 바꿔줄 필요 없이 하루동안 쌓인 데이터를 조회할 수 있었다. 그런데 이와는 달리 복잡한 경우가 있다. dag이 매월 말일(0 0 L * *)에 도는 스케줄을 가지고 있을 때 전월 말일부터 1일 전까지의 데이터만 조회하고 싶을 때이다. SELECT NAME, ADDRESS FROM TBL_REG WHERE REG_DATE BE..