개발자 노트

Spark 으로 ETL 작업을 처리하다가 처리한 데이터의 누적 양을 집계하고 싶었다. 예를 들면, Kafka의 데이터를 활용해 spark streamming 작업을 할 때, 각 단계(spark streaming은 짧은 간격의 배치)마다 처리된 데이터 건수를 집계한다고 하자. spark은 분산으로 데이터를 처리하기 때문에 각 executor의 처리된 결과를 조합한 공유 변수가 필요하다. executor가 처리한 데이터를 하나의 공유변수로 값을 기록하는 것이다. 이 때 공유변수는 결합 및 가환 연산을 지원해야 한다. '가환'이라는 단어가 어색한데 사전으로 찾아본 결과 '조작이나 연산의 순서를 바꾸어도 그 결과가 변하지 않는 일' 이다. 다시 간단히 정리하면 공유변수를 통한 연산의 순서, 연산의 조합의 결과가..

최근에 Kafka 인증 기술에 대해서 하나씩 공부를 하고 있다. 가장 기본이 되는 SASL/PLAIN에 대해서 이전 블로그 글을 통해 소개한 바 있다. 이번 글은 Kafka 2.0 버전에 추가된 SASL callback 기능을 알아보고자 한다. SASL callback이 인증과 관련하여 중요하다고 생각되는 것은 SASL/PLAIN에서 접근을 허용하는 사용자 정보의 관리에 장점이 있어서이다. SASL/PLAIN에서 소개했듯이 Broker의 jaas 파일에 사용자 정보를 관리하게 되면 Kafka 운영에 불편한 점이 많다. 이번에 소개할 SASL callback 기능을 사용하면 인증 로직을 custom하게 추가할 수 있어 Kafka Cluster 운영 상황에 맞는 인증 서비스를 할 수 있다. KIP-86에 S..

예전 블로그 글 - [Kafka] Topic의 Replica 분배 로직 분석을 통해 Kafka는 Topic을 만들 때 각 Broker에 Partition을 균일하게 분배한다고 설명한 적이 있다. 하지만 Kafka를 운영하면 각 Topic 별로 쌓이는 데이터 용량이 다르기 때문에 Broker별 디스트 사용량이 차이가 나게 된다. 필자도 회사 업무 중에 디스크 불균형이 발생하여 일부 Topic의 Partition을 디스크가 남은 Broker로 이동하는 작업을 진행하였다. 먼저 어떤 Topic들을 대상으로 이동하면 좋을지 선택해야 했다. 모니터링 metric을 수집하기 때문에, 수집된 metric으로 판단을 해도 좋다. 에를들면, kafka.server:type=BrokerTopicMetrics,name=By..

최근 Kafka 인증에 대해서 살펴보고 있는데, 인증에서 가장 기본이 되는 SASL/PLAIN에 대해서 정리해 보고자 한다. 블로그의 내용은 Kafka Document의 내용을 토대로 직접 실습해보고 정리한 내용이다. SASL/PLAIN 인증 방식은 가장 먼저 머리속에 떠오르는 id/password를 통한 인증이다. id/password로 인증을 할 때는, SSL 통신을 해야 plain text가 노출되지 않는다. Kafka에서는 SASL_SSL, SASL_PLAIN 2가지 타입이 있으며, Live 환경에서는 SASL_SSL으로 설정해서 SSL 통신을 하도록 추천하고 있다. SASL/PLAIN 인증을 하기 위해서는 인증정보를 포함한 conf 파일을 Broker, Client에 각각 만들어 줘야 한다. 먼..

회사에서 서비스를 만들다가 Restful API를 통해 파일을 업로드 받고 HDFS에 저장하는 기능이 필요했다. API 서버는 Spring으로 구현되어 있었기에 Spring에서 Hadoop 관련 repository도 찾아보고 JavaConfig 설정도 해보았는데 이번 포스트에서 간단히 정리해 보고자 한다. 내용은 아래의 순서로 진행하고자 한다. 1. Spring Apache Hadoop 라이브러리 찾기. 2. Hadoop 관련 JavaConfig 추가. 3. Hadoop NameNode HA 구성 맛보기. 4. HDFS에 파일 업로드. 1. Spring apache hadoop 라이브러리 찾기. Spring Data 프로젝트 하위에 Spring for Apache Hadoop이 포함되어 있다. 라이브러리..

Kafka consumer 관련 2번째 글이다. 지난 글에서 LAG에 대해서 설명했는데, consumer group과 offset에 대한 설명을 빼 놓고 지나가니 이해하는데 부족할 수 있어 consumer group에 대한 개념 설명을 하고자 한다. 그리고 추가로 consumer group의 offset을 변경하고자 할 때가 발생할 수 있는데 간단한 python 코드로 offset을 변경하는 방법을 알아보자. offset을 왜 변경하지? 라는 질문이 있을 수 있는데, consumer group에 대해서 이해하면 질문에 대한 답이 저절로 될 듯 하다. 1. consumer group 아래 그림이 consumer group을 이해할 수 있는 가장 좋은 그림이다. Kafka의 Topic은 여러 partitio..

Kafka에 저장된 데이터를 consumer를 통해 처리를 하고 있을 때, 개발한 코드로 데이터가 잘 처리되고 있는 건지 궁금할 때가 있다. consumer의 데이터 처리 상황을 주기적으로 모니터링 한다면 처리하지 못하고 Kafka에 남아있는 데이터 수치를 알아야 한다. Kafka에 남아 있는 데이터 수의 증감을 통해 consumer의 throughput도 계산할 수 있다. Kafka 문서에 보면 consumer group의 현재 position을 얻는 command가 있다. 다음은 해당 command인 kafka-consumer-group.sh의 실행 예제이다. > bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --g..

Kafka에서 topic을 생성할 때, partition 별로 replica를 어떻게 할당하나 찾아보았다. 여러개의 broker를 가진 Kafka Cluster를 운영할 경우 각 broekr 마다 replica 불균형이 발생한다. 최고 topic 생성 이후에 broker shutdown 등 다른 이유에 의해 replica가 바뀔 수 있겠지만 topic 생성 최초에 partition 별 replica 할당이 어떻게 되나 궁금했기 때문이다. 그래서 topic이 생성하는 코드 안쪽에서 partition 분배 로직을 뒤졌다. partition 분배 로직에는 broker의 rack 설정이 있는 것과 없는 것 2개의 로직이 존재했다. 이 글에서는 rack이 없는 설정에 대해서 먼저 분석해 본다. 다음은 partit..

팀에서 spark SQL을 사용해서 사용자가 요청한 쿼리를 stream 처리하는 작업을 하고 있다. 그래서 spark SQL에 대해서 정리해 보고자 한다. 해당 블로그의 내용은 책 '스파크 2 프로그래밍'에서 대부분 참고했다. 1. RDD와의 차이점 spark의 기본 데이터 모델은 RDD 이다. 분산환경에서 메모리 기반으로 빠르고 안정적으로 동작하는 프로그램을 작성할 수 있는 장점이 있지만 아쉬운 점도 있었다. 그것은 "스키마"에 대한 표현방법이 없다는 것이다. spark 2.0 부터 DataSet으로 통합되어 "스키마" 있는 데이터 처리가 가능하다. "스키마"라는 단어를 처음 들었을 때, 제일 먼저 데이터베이스가 생각났다. 데이터베이스가 설정된 스키마에 따라 SQL문을 던지면 결과를 주듯이, DataS..