데브원영

카프카 커넥트(분산모드)는 기본 JMX포트를 통해 커넥트 애플리케이션의 상태를 모니터링 수집 애플리케이션에 데이터를 전송할 수 있습니다. 오늘은 카프카 커넥트의 JMX포트를 열고 로그스태시로 JMX로 지표를 수집하는 방법에 대해 알아보겠습니다. 1. 카프카 커넥트 JMX포트 열기 $ export JMX_PORT=10000 우선 JMX포트를 열기위해 환경변수로 선언합니다. 2. 카프카 커넥트 실행 및 JMX 포트 확인 $ bin/connect-distributed.sh ../config/connect-distributed.properties 실행이 완료되면 설정한 10000번 포트로 listen 여부를 확인합니다. $ netstat -anv | grep 10000 tcp46 0 0 *.10000 *.* L..

로컬에서 테스트 용도로 잠깐씩 엘라스티서치, 키바나를 도커로 띄울 니즈가 있을때가 있습니다. docker와 docker-compose를 활용하여 로컬에서 띄우는 방법을 알아봅니다. 1. 도커설치 $ brew cask install docker 도커 설치는 homebrew를 활용합니다. 2. docker-compose.yml 파일 생성 version: '3' services: elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:6.3.2 environment: - cluster.name=docker-cluster - bootstrap.memory_lock=true - "ES_JAVA_OPTS=-Xms512m -Xmx512m" ulimit..

카프카 커넥트를 운영하다보면 로그를 파일로 쌓고 + 로그스태시를 통해 엘라스틱서치에 쌓고 싶을때가 있습니다. 이를 위해 아래와 같이 설정합니다. 1. connect-log4j.properties router.logs.dir=./connect-logs log4j.rootLogger=INFO, connectAppender log4j.logger.org.apache.zookeeper=ERROR log4j.logger.org.I0Itec.zkclient=ERROR log4j.logger.org.reflections=ERROR log4j.appender.connectAppender=org.apache.log4j.DailyRollingFileAppender log4j.appender.connectAppender.F..

인증서(Certificate) 디지털 인증서라고도 부른다. 유저 -> 브라우저 또는 서버 -> 서버 간에 암호화된 연결을 수립하는데 사용된다. 인증되지 않는 사용자의 네트워크 데이터 탈취를 막기 위한 용도. SSL 인증서는 공개키 방식을 사용한다. 공개키 방식은 공개키와 비밀키 방식이 있다. 공개키 방식으로 암호화하고 비공개키로 복호화한다. 비공개키가 없으면 복호화할 수 없다. CA(Certificate authority) 인증서를 보장하는 기업들을 CA 또는 Root Certificate라고 부른다. CA는 신뢰성이 업격하게 공인된 기업들만이 참여할 수 있다. 대표적인 기업으로는 Symantec, Comodo, GlobalSign 등이 있다. 암호화된 통신을 제공하려는 서비스는 CA를 통해서 인증서를 ..

openjdk 설치 [ec2-user@ip-172-31-6-79 ~]$ sudo yum install -y java-1.8.0-openjdk-devel.x86_64 Loaded plugins: extras_suggestions, langpacks, priorities, update-motd Resolving Dependencies --> Running transaction check ---> Package java-1.8.0-openjdk-devel.x86_64 1:1.8.0.252.b09-2.amzn2.0.1 will be installed --> Processing Dependency: java-1.8.0-openjdk(x86-64) = 1:1.8.0.252.b09-2.amzn2.0.1 for pac..

log4j는 properties파일을 통해 로깅 파일을 어떻게 활용할 것인지 선언할 수 있습니다. 그런데 이 파일에서 DailyRollingFileAppender를 사용할 경우 순서를 반드시 지켜야합니다. 그 순서란 아래와 같습니다. log4j.appender.outAppender=org.apache.log4j.DailyRollingFileAppender log4j.appender.outAppender.DatePattern='.'yyyy-MM-dd log4j.appender.outAppender.File=${router.logs.dir}/application.log log4j.appender.outAppender.layout=org.apache.log4j.EnhancedPatternLayout log4j..

첫번째는 데이터 적재의 병목현상이다. 서비스의 사용자가 많아지고 애플리케이션의 복잡도가 증가하면 생성 되는 데이터 양도 기하급수적으로 늘어난다. 데이터를 데이터 레이크로 즉시 보내려고 하지만 데이터 레이크로 바로 적재하는 것은 병목현상이 발생할 수 밖에 없다. 또한 엔드 투 엔드방식의 적재로 인해 데이터 레이크로 사용하는 저장소에 장애가 발생할 경우 서비스에 영향도가 전파되어 서비스가 중단되거나 장애가 발생한다면 사용자에게 불편한 경험을 줄 수 있다. 두번째는 다양한 저장소 이슈이다. 데이터 레이크 저장도구로서 가장 널리 사용되는 프레임 워크는 하둡(Hadoop)이다. 하둡은 구글(Google)이 논문으로 발표한 GFS(Google File System)와 맵리듀스(MapReduce)를 구현한 오픈소스이..

1. centos6 AMI로 instance 생성 2. root권한 획득 및 유저 변경 $ sudo passwd Changing password for user root. New password: Retype new password: passwd: all authentication tokens updated successfully. $ su root Password: 3. 관련 패키지 설치 $ yum install wget gcc gcc-c++ git 4. golang 다운로드 $ wget https://dl.google.com/go/go1.15.linux-amd64.tar.gz $ tar -C /usr/local -xvzf go1.15.linux-amd64.tar.gz 5. 환경변수 설정 $ vi /e..

1. homebrew를 통해 hadoop package 설치 $ brew install hadoop hadoop이 설치된 경로는 아래와 같이 명령하여 확인할 수 있다. $ brew info hadoop hadoop: stable 3.3.0 Framework for distributed processing of large data sets https://hadoop.apache.org/ Conflicts with: yarn (because both install `yarn` binaries) /usr/local/Cellar/hadoop/hdfs (366 files, 40.9MB) Built from source /usr/local/Cellar/hadoop/3.3.0 (21,844 files, 963.5MB..

EMR 노트북에서 pyspark를 실행하는데 암/복호화를 위해 secret manager에 암호화 키를 저장하고 조회해야하는 니즈가 있었습니다. 이를 해결하기 위해 secret manager를 사용하였고 어떤 방식으로 진행했는지 알아보겠습니다. 암호화 키를 코드에 직접 hard coding하는 방식은 보안에 매우 취약합니다. AWS를 사용하는 이상 AWS관련 컴포넌트를 활용하여 비밀 키/값을 관리하고 이를 EMR notebook에서 활용하는 방법을 알아보겠습니다. 관련 AWS 컴포넌트 - EMR - EMR notebook - AMI - secret manager EMR 생성시 EC2 인스턴스 프로파일 설정 EMR notebook은 서버리스 주피터/주피터랩입니다. EMR notebook에서 pyspark,..

JVM기반 애플리케이션을 개발하다가 이슈가 생겼을때 다양한 방식으로 이슈를 확인하고 처리할 수 있습니다. 그중 한가지인 스레드 덤프를 사용하여 이슈를 해결한 사례를 공유하겠습니다. 제가 운영하고 있는 애플리케이션은 Scala로 이루어져 있고 JDK 1.8에서 돌아가는 Multithread application입니다. 카프카 컨슈머를 스레드로 작게는 10개 많게는 100개이상 띄워서 운영하고 있습니다. 이 애플리케이션은 안전하게 중지하기 위해 Shutdown hook을 받도록 개발했습니다만, 언제부턴가 Shutdownhook을 통해 중지요청을 하더라도 애플리케이션이 중지하는 절차를 진행하지 않는 현상이 발생했습니다. 또다시 shutdown hook을 kill -term 명령어와 함께 명령내렸으나 응답이 없..

AWS EMR은 하둡관련 패키지를 AWS를 통해 설치하고 프로비져닝하는 도구입니다. 이 도구를 활용하면 매우 빠르게 hadoop, spark 등의 빅데이터 클러스터를 구축할 수 있는데요. 오늘은 이렇게 만든 EMR 클러스터에서 3rd party python library사용을 위한 설치 스크립트를 적용해보겠습니다. 저같은 경우는 pyspark를 자주 사용하는데요. pyspark사용할 경우 추가 pip3라이브러리를 사용해야할 때가 있습니다. pip3 라이브러리를 pyspark로 실행할 경우 스파크가 배포되는 모든 인스턴스에 해당 라이브러리가 설치되어 있어야합니다. EMR에는 bootstrap action이라고 하는 인스턴스 생성할때 실행할 수 있는 스크립트를 지정할 수 있습니다. 여기에 스크립트를 추가하여..

AWS에서 VPC지정을 위해 내부 사설 IP주소대역을 CIDR방식으로 지정합니다. CIDR방식을 이해하지 못하면 내부에 지정되는 사설 IP값이 어떻게 지정되는지 알지 못합니다. 이번 포스팅에서는 CIDR계산방법을 알아보겠습니다. CIDR 사이더(Classless Inter-Domain Routing, CIDR)는 클래스 없는 도메인 간 라우팅 기법으로 1993년 도입되기 시작한, 최신의 IP 주소 할당방법입니다. IPv4를 내부적으로 효율적으로 사용할 수 있도록 지정할 수 있습니다. /를 기준으로 앞은 host ID, 뒤는 network ID가 됩니다. 그런데 주의할 점은 CIDR의 / 뒤쪽 숫자를 잘못 적으면 올바른 CIDR이 아닐 수 있으므로 주의해야합니다. 예제 10.0.1.0/24 앞쪽 24비트를..

스프링 카프카에서 메시지 리스너를 구현하는 2가지 방법이 있습니다. 첫번째는 @KafkaListener를 사용하는것이고 두번째는 listenerContainer를 Bean등록하는 것입니다. @KafkaListener로 구현 가장 간단한 방법입니다. @KafkaListener를 통해 선언할 경우 파라미터를 오버로딩해서 알맞는 listenerContainer를 자동으로 주입합니다. package com.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBoot..

카프카 컨슈머는 토픽의 파티션과 매칭하여 레코드를 가져옵니다. 파티션을 매칭하는 기준은 컨슈머 파티션 어사이너의 기준을 따릅니다. 컨슈머 파티션어사이너 인터페이스는 아래 링크에서 확인할 수 있습니다. https://kafka.apache.org/25/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html ConsumerPartitionAssignor (kafka 2.5.0 API) kafka.apache.org RangeAssignor 기본 설정되는 파티션 어사이너입니다. 토픽의 파티션을 숫자기준으로 나열하고, 컨슈머의 이름을 사전순으로 나열한 뒤에 배정하는 정확히 반으로 나누어 배정합니다. 만약 딱 반으로 안나뉘어지는 홀수개의 ..

카프카 프로듀서는 레코드를 전송하기 위해 파티셔너를 제공합니다. 파티셔너 종류와 각 파티셔너에 대한 설명을 정리해보았습니다. 카프카 버젼이 올라감에 따라 파티셔너의 종류도 달라졌는데 여기서는 2.5.0 버젼의 파티셔너를 정리해보겠습니다. 프로듀서 파티셔너 인터페이스에 대한 설명은 아래 링크에서 확인할 수 있습니다. https://kafka.apache.org/25/javadoc/?org/apache/kafka/clients/producer/Partitioner.html kafka 2.5.0 API kafka.apache.org 메시지 키가 있을 때 메시지 키가 있을 경우에는 파티셔너의 종류와 관계없이 동일하게 동작합니다. 메시지 키의 해쉬값을 구해서 해당 해쉬값과 파티션을 매칭하여 적재합니다. 이로 인해..

오늘 카프카 컨슈머를 입수하던 도중 매우 신기한 현상을 발견해서 기록하게 되었습니다. 컨슈머스레드를 여러개로 HDFS에 적재하고 있었는데, 일부 파티션의 데이터가 0이 되어 일부 컨슈머 스레드가 쉬는동안 총 처리량이 늘어난 것입니다. 그래프를 보면서 왜 그렇게 되었는지 확인해보겠습니다. 입수 상태 - 24 core, 48G memory 물리장비 - 1 JVM application - 20 consumer thread - sync offset commit - save at Hadoop hdfs(append) 입수 관련 그래프 컨슈머 랙은 카프카 버로우를 사용하여 수집중에 있습니다. 랙 추이를 보면 알겠지만 파티션당 데이터 양의 차이가 일부 발생하였습니다. 10개 파티션은 18:20 부근에 끝났고 나머지 1..

저번 포스트에 이어 distributed로 동작하는 kafka connector를 실습해보겠습니다. https://blog.voidmainvoid.net/356 Kafka file source connector standalone 모드 실행 Kafka에는 커낵터가 있습니다. 다양한 커낵터 클래스를 사용하여 컨슈머나 프로듀서 작성 없이 source로 부터 데이터를 카프카로 보내거나 받을 수 있습니다. 오늘은 FileStreamSource를 사용해서 file을 blog.voidmainvoid.net 준비물 - 카프카 바이너리 - 실행중인 카프카 클러스터 distributed vs standalone The difference is in the class which is started and the confi..

Kafka에는 커낵터가 있습니다. 다양한 커낵터 클래스를 사용하여 컨슈머나 프로듀서 작성 없이 source로 부터 데이터를 카프카로 보내거나 받을 수 있습니다. 오늘은 FileStreamSource를 사용해서 file을 카프카로 보내보겠습니다. 준비물 - 카프카 바이너리파일 - 카프카 클러스터 Standalone 모드 준비 Standalone모드란 1개의 프로세스로 동작하는 애플리케이션을 뜻합니다. 커낵터는 distribute모드와 standalone모드를 지원합니다. 실제 환경에서는 distribute모드를 사용하는 것을 권장합니다. standalone모드를 실행하기 위해 아래와 같이 준비합니다. connect-file-source.properties name=local-file-source conne..

mac OS에서 homebrew를 통해 tree명령어를 설치하면 디렉토리를 ascii tree로 보여줍니다. $ brew install tree $ tree . ├── build.gradle ├── gradle │ └── wrapper │ ├── gradle-wrapper.jar │ └── gradle-wrapper.properties ├── gradlew ├── gradlew.bat ├── settings.gradle └── src ├── main │ ├── java │ └── resources └── test ├── java └── resources 9 directories, 6 files

kafka-client를 java application에 추가하여 실행하면 아래와 같은 오류와 함께 log가 정상적으로 찍히지 않습니다. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 위 에러는 slf4j의 디펜던시 이슈로 StaticLoggerBinder가 없어서 그런것인데요. 아래와 같이 그래들에 디펜던시를 추가하면 로그를 확인하실 수 있습니다. dependencies { co..

EMR노트북은 이미 생성된 EMR클러스터를 활용할 수 있게 만든 주피터 노트북입니다. EMR클러스터에 직접 인스턴스 접속하여 각종 빅데이터 처리를 해도되지만 주피터를 사용한다면 좀더 편리한 환경에서 R, Python 등을 사용할 수 있습니다. 오늘은 이전에 생성한 스파크 EMR클러스터에 노트북을 연결하여 사용해보는 시간을 가지도록 하겠습니다. 스파크만 가지고 있는 소규모 EMR클러스터 생성은 아래 링크에서 확인하실 수 있습니다. https://blog.voidmainvoid.net/351 소규모 스파크 사용을 위한 AWS EMR 클러스터 생성하기 AWS에는 EMR이라고 불리는 빅데이터 플랫폼이 있습니다. EMR을 통해 스파크, 하이프, HBASE, 플링크 등 다양한 오픈소스 도구 셋트를 생성할 수 있습니..

AWS에는 EMR이라고 불리는 빅데이터 플랫폼이 있습니다. EMR을 통해 스파크, 하이프, HBASE, 플링크 등 다양한 오픈소스 도구 셋트를 생성할 수 있습니다. 온프로미스로 직접 구축하는 것에 비해 매우 빠른 속도로 구축할 수 있으며 AWS에 따르면 온프로미스에서 구축하는 비용에 50%로 사용할 수 있다고 합니다. 오늘은 S3에 저장된 데이터를 스파크로 처리하기 위해 소규모 스파크로만 구성된 EMR클러스터를 생성해보겠습니다. 준비물 - AWS계정 S3 저장소 생성 S3 저장소를 생성하는 것은 쉽습니다. S3 콘솔화면으로 가셔서 버킷을 신규로 생성하면 됩니다. EMR 클러스터 생성 emr클러스터는 aws의 emr콘솔 화면에서 생성할 수 있습니다. 콘솔로 가셔서 아래와 같은 단계를 따라 하시면 됩니다. ..

카프카의 ConsumerRecord를 살펴보면 메시지 키와 메시지 값 이외에 추가로 timestamp가 있는 것을 확인할 수 있습니다. timestamp는 Kafka 프로듀서가 ProducerRecord를 생성하면서 timestamp 값을 Unix time으로 입력합니다. 만약 다른 timestamp를 넣고 싶다면 직접 설정하여 넣을수도 있습니다. 해당 건은 KAFKA-3025에서 논의했던 내용으로 자세한 내용은 해당 jira에서 확인할 수 있습니다. https://issues.apache.org/jira/browse/KAFKA-3025 [KAFKA-3025] KIP-31&KIP-32 (part 1): Add timestamp field to message, configs, and Producer/Con..

telegraf는 metric수집에 최적화되어있습니다. 다만, 그대로 사용하게 되면 influxdb에서 사용하는 데이터형태(csv와 유사)로 데이터가 전송되는데요. kafka에서 데이터를 잘 활용하기 위해 json을 사용한다면 consumer에서도 스키마에 대한 걱정없이(?) 잘 사용할 수 있습니다. telegraf에서 data형태를 json형태로 바꾸는 예제를 아래에 설명드리도록 하겠습니다. telegraf에 대한 설명은 아래 링크를 확인해주세요. https://blog.voidmainvoid.net/260 Fluentd vs Telegraf 차이점 알아보기 Telegraf와 fluentd는 아주 유사해보인다. 둘다 configuration파일 기반으로 작동하며 plugin을 통해 개발자가 custom..

pyspark를 사용하여 dataframe을 처리한 이후 csv로 json value를 저장할때 아래와 같이 저장될 때가 있습니다. // 저장되기 원하는 값 {"key":"value"} // 실제로 저장되는 값 "{\"key\":\"value\"}" json value로 그대로 저장하길 원하지만, 따옴표가 escape처리되어 저장됩니다. 이때는 pyspark의 csv저장옵션을 지정해야합니다. pyspark csv 저장옵션 중 escapeQuotes를 false로 두면 해결할 수 있습니다. ... dataframe.write.option("escapeQuotes","false").mode("overwrite").csv("/data/destination") 위 구문을 실행하면 json이 그대로 저장되는것을 ..

스파크를 다루다보면 jupyterlab을 통해 pyspark를 실행시킬 일이 종종 있습니다. 로컬컴퓨터(macOS)에 설치하는 방법을 설명드리도록 하겠습니다. pyspark와 jupyterlab을 설치하기 위해서 패키지 설치도구인 homebrew를 사용하도록 하겠습니다. homebrew에 대한 자세한 설명은 아래 영상을 참고해주세요. 1. pyspark설치 $ brew install apache-spark 2. jupyterlab설치 $ brew install jupyterlab 3. pyspark환경변수 설정 pyspark를 실행할때 아래 2개의 환경변수를 .bashrc 또는 .zshrc에 넣어도록합니다. pyspark를 실행하면 jupyterlab을 실행하게 됩니다. export PYSPARK_DRI..

지난 6월 19일(금) SK플래닛 T아카데미(서울대연구공원 단지 내 SK연구동)에서 '아파치 카프카 입문과 활용'이라는 제목으로 카프카강의를 수행하였습니다. 총 강의 시간은 5시간으로서 오후 1시부터 시작하여 오후 6시에 끝나는 꽤나 긴 시간의 강의였습니다. 오늘 포스트에서는 제가 기술 강의를 수행하기 위해 어떤 준비를 했는지, 어떤 과정을 거치고 어려운점은 무엇이었는지, 마지막으로 느낀점에 대해서 정리하겠습니다. 강의를 제안받다 제가 초청받은 강의는 토크ON세미나입니다. SK플래닛 T아카데미에서 주최하는 이 세미나는 ICT분야의 커리어 설계를 준비하는 대학생을 위한 Interactive세미나로서, 다양한 ICT주제로 매달 세미나를 열고 있습니다. SK플래닛의 사내/외 전문가를 초청하여 세미나를 열고 있..

confluent에서는 rest proxy라고 불리는 카프카 클러스터를 위한 RESTful interface application을 오픈소스로 제공하고 있습니다. 기존에 Kafka connect, Kafka client로 데이터를 전달하는 것과는 사뭇 다르게 REST api를 사용한다는점이 독특한데요. 직접 코드를 짜지 않고 범용적으로 사용되는 http을 사용해서 데이터를 넣고 뺄 수 있다는 점이 독특합니다. 오늘은 rest proxy를 local에 설치하고 실행해보겠습니다. 준비물 - local kafka cluster - git - terminal - postman 다운로드 및 실행 rest proxy를 사용하기 위해서는 rest proxy가 포함된 confluent의 community package..

AWS MSK는 카프카를 SaaS형태로 사용가능한 AWS서비스중 하나입니다. 완전관리형 SaaS는 아니지만 다양한 옵션을 지정할 수 있으며, 추가적인 모니터링 도구를 제공해주기 때문에 아주 편리하게 사용할 수 있습니다. MSK에서는 총 7개의 인스턴스유형을 제공하는데, 각 인스턴스 유형별로 생성할 수 있는 토픽의 개수를 살펴보도록 하겠습니다. MSK Bestcase 문서에 따르면 적정크기의 클러스터에 대한 내용이 나와 있습니다. 여기서는 MSK클러스터를 생성할 때 브로커 노드의 유형 및 수에 대한 상세한 내용이 나와 있는데 그 중 브로커당 파티션수에 대한 내용이 아래와 같이 정리되어 있습니다. 위 그림에서 보는 것 처럼 t3.small이 가장 작고 m5.24xlarge가 가장 큰 파티션 개수를 가질 수 ..