데브원영

kafka-lag-dashboard Kafka lag을 모니터링하는 확실한 방법 Kafka Consumer의 처리시간이 지연되면 topic 내부의 partition lag이 증가합니다. lag 모니터링을 통해 어느 partition이 lag이 증가하고 있는지, 어느 컨슈머가 문제가 있는지 확인하기 위해서는 consumer단위의 metric 모니터링으로는 해결하기 쉽지 않습니다. 그렇기 때문에 카프카 컨슈머 모니터링을 위해서는 burrow와 같은 외부 모니터링 tool 사용을 권장합니다. 이 문서에서는 Linkedin에서 제공한 burrow를 사용하여 lag정보를 Elasticsearch로 수집하는 데이터파이프라인을 만들어보고, Grafana 기반의 consumer단위 lag 모니터링 대시보드를 만드는 방..

Kafka client 2.0부터는 KafkaConsumer.poll(long timoutMs)는 deprecated되었다. KafkaConsumer.poll(long timeoutMs)를 기존처럼 long type parameter로 사용할 경우 poll(Duration timout)으로 redirect된다. 이 수정사항은 KIP-266에 의해서 수정되었고 수정된 사유를 아래와 같이 적어보고자 한다. KafkaConsumer.poll(long) poll method는 consumer에서 빠트릴 수 없는 중요한 메서드이다. 데이터를 가져오는 역할을 하며 무한루프안에서 지속적으로 호출되어 topic으로부터 데이터를 가져오는데 사용된다. 기존에 사용되던 poll() method는 long type 파라미터로..

이번영상에서는 아파치 카프카 토픽에 대해서 알아보았습니다. 카프카 토픽은 데이터가 저장되는 곳인데요. 토픽은 카프카에서 가장 기본적이고 필수로 알아야 합니다.

빅데이터🤩에서 가장 많이 쓰이는 아파치 카프카에 대해서 알아보았습니다. 카프카는 정말 많은 기업에서 빅데이터처리에 사용하고 있습니다.

KSQL - 효과적이고 간편한 스트리밍 프로세스 SQL엔진 from Won young Choi KSQL은 스트리밍 application을 SQL 쿼리를 사용하여 만들 수 있습니다. KSQL에 대한 전반적인 내용을 슬라이드에 담아보았습니다. - Kafka environment - KSQL Examples - KSQL and stream - KSQL format - KSQL data types - KSQL architecture - KSQL 하위호환성 - Custom KSQL(UDF, UDAF) - Who need KSQL - KSQL License

KSQL에서 stream을 생성하다가 아래와 같은 오류를 만났다. line 3:3: extraneous input 'properties' expecting {'INTEGER', 'DATE', 'TIME', 'TIMESTAMP', 'INTERVAL', 'YEAR', 'MONTH', 'DAY', 'HOUR', 'MINUTE', 'SECOND', 'ZONE', 'PARTITION', 'STRUCT', 'EXPLAIN', 'ANALYZE', 'TYPE', 'SHOW', 'TABLES', 'COLUMNS', 'COLUMN', 'PARTITIONS', 'FUNCTIONS', 'FUNCTION', 'ARRAY', 'MAP', 'SET', 'RESET', 'SESSION', 'IF', IDENTIFIER, DIGIT_..

아래 포스트는 엔지니어 블로그를 번역하였습니다. 2019년 10월 8일 Linkedin이 엔지니어 블로그를 통해 내부에서 사용중인 kakfa를 공개하였다. blog : https://engineering.linkedin.com/blog/2019/apache-kafka-trillion-messages github : https://github.com/linkedin/kafka 아파치 카프카는 링크드인의 인프라에서 중요한 부분을 차지한다. 아파치 카프카는 처음에 in-house 스트림 프로세싱 플랫폼으로 개발하다가 오픈소스로 공개하였고 오늘날에는 많은 적용사례들이 있다. 아파치 카프카는 activity tracking, message exchange, metric gathering과 같은 역할을 하는 소프트..

아래 포스트는 airbnb의 kafka meetup 발표내용을 토대로 정리하였습니다. youtube url : https://www.youtube.com/watch?v=-PmKeOpfE54 Kafka at Airbnb AWS에서 0.11 버젼의 kafka를 사용하고 있으며 10개 이상의 cluster과 500개 이상의 broker로 구성되어 있다. 3000개 이상의 topic들이 사용되어지고 있으며 모든 데이터의 양의 PB단위 이상이다. 아직 0.11버전이지만 다음달(2019년 11월)에는 그 이상의 버젼을 사용할 것이라고 한다. 위 데이터 파이프라인 아키텍쳐에서 Rest Proxy는 여러 language로 작성된 application에서 나온 데이터를 kafka로 prodcue하기 위한 역할을 하는데 ..

KSQL은 스트리밍 application을 SQL 쿼리를 사용하여 만들 수 있다. KSQL은 Kafka stream으로 만들어져 있다. KSQL은 Kafka 클러스터와 연동되는데 이는 기본적인 Kafka stream application동작구조와 동일하다. KSQL 아키텍쳐와 주변 application들 KSQL은 아래와 같은 구성요소로 이루어져 있다. KSQL 아키텍쳐 - KSQL 엔진 : KSQL 쿼리가 실행되고 있는 곳 - REST 인터페이스 : ksql 엔진에 client로 access할 수 있는 인터페이스 주변 application - KSQL CLI : KSQL 엔진에 CLI(Command Line Interface)로 접속할 수 있게 도와주는 application - KSQL UI : Con..

아파치 카프카는 대규모 분산 스트리밍 플랫폼으로서 데이터파이프라인을 만들때 주로 사용이 가능하다. 데이터파이프라인을 만듦에 있어서 어떤 용도로 어떻게 동작하는지 확인하기 위해서는 직접 consumer로 데이터를 넣어주어 producer을 개발하거나 혹은 처음엔 cli producer/consumer을 사용하여 data를 topic에 넣어준다. Confluent사에서 제공하는 ksql-datagen을 사용한다면 여러 format으로 data를 auto generate하여 topic에 produce가능하다. ksql-datagen은 ksql을 설명하기 위해 처음 소개되었지만, ksql뿐만 아니라 kafka에서 consumer을 테스트하거나 다양한 format(avro 등)을 테스트하기에도 알맞다. 이번 포스..

KSQL은 SQL을 사용하여 Kafka topic으로 들어오는 record들에 대해 query문을 작성하여 transform하거나 aggregation등을 수행할 수 있게 만들어준다. KSQL을 작성하고 사용하기 위해서는 KSQL 서버가 반드시 필요한데, 이번 포스팅에서는 KSQL 서버를 Docker를 사용하여 설치 및 실행해보고자 한다. 1. KSQL 서버 설치 및 실행 본인의 macbook에서 docker image로 실행하기 때문에 아래와 같이 command를 실행한다. $ docker run \ -p 127.0.0.1:8088:8088 \ -e KSQL_BOOTSTRAP_SERVERS=localhost:9092 \ -e KSQL_LISTENERS=http://0.0.0.0:8088/ \ -e KS..

출처 : slideshare-Kafka Streams vs. KSQL for Stream Processing on top of Apache Kafka Kafka는 Bigdata를 처리하고 운영함에 있어서 필수불가결하다. 이미 많은 IT기업들(카카오, 네이버 등)에서는 kafka로 동작하는 실서비스를 운영하고 있으며 그에 대한 know-how도 상당히 많이 공유되고 있다. Kafka는 단순히 produce, consumer 구조로 사용가능하지만, KSQL이나 kafka stream을 사용하여 더욱 효과적이고 유연하게 데이터를 조작할 수 있다. 이번 포스팅에서는 KSQL과 Kafka stream에 대해서 소개하고 차이점도 알아보는 시간을 가지려고 한다. KSQL KSQL은 streaming SQL 엔진으로서..

아래 포스팅은 DZone의 What Is a Data Pipeline? 을 번역하였습니다. url : https://dzone.com/articles/what-is-a-data-pipeline 이번 포스팅에서는 데이터파이프라인에 대해서 정의하고 데이터파이프라인을 통해 추구하는 바를 알아보고자 한다. 데이터를 효과적으로 가져오는 것은 오늘날 Data-driven enterprise의 큰 숙제이다. 데이터를 어느 한 지점에서 특정공간까지 가져오는 데에 많은 장애물들(중복 유실 등)이 있으며 가져오려는 지점이 많아지면 많아질수록 더욱 문제가 복잡해진다. 데이터 파이프라인 : 효율을 위한 작업 데이터 파이프라인의 시작은 왜, 어디에서, 어떻게 데이터를 수집할 것인가에서 부터 시작한다. 데이터 파이프라인을 구축하..

아래는 oreilly의 Kafka: The Definitive Guide(카프카 핵심가이드)의 commitSync()와 rebalanceListener를 사용하여 topic을 consume하는 예시 코드이다. url : https://www.oreilly.com/library/view/kafka-the-definitive/9781491936153/ch04.html private Map currentOffsets = new HashMap(); private class HandleRebalance implements ConsumerRebalanceListener { public void onPartitionsAssigned(Collection partitions) { } public void onPartit..

Github의 awslabs에서는 여러 aws관련(혹은 기타)관련 repository들을 opensource로 운영중에 있다. - awslabs github url : https://github.com/awslabs 여기에 특별한, 보물같은 spark라이브러리를 최근에 facebook을 통해 찾았는데, 이름은 Deequ이다. - deequ github url : https://github.com/awslabs/deequ Deequ Deequ는 Apache Spark에서 돌아갈 수 있는 large dataset에 대해서 data기반의 unit test를 하여 data quality를 측정할 수 있는 라이브러리이다. Deequ 적용 방법 Deequ는 Java8에서 Spark 2.2.x~2.4.x 에서 동작하..

https://books.google.co.kr/books?id=a3wzDwAAQBAJ&pg=PA77&lpg=PA77 Kafka: The Definitive Guide Every enterprise application creates data, whether it’s log messages, metrics, user activity, outgoing messages, or something else. And how to move all of this data becomes nearly as important as the data itself. If you’re an application architect, develop books.google.co.jp 참고 출처 - Kafka Definitive gui..

Fluentd개요 fluentd는 대용량 데이터처리에 있어 input/output plugin들을 사용해서 파이프라인을 생성할 수 있다. 이 파이프라인은 데이터처리에 적합한데 다양한 플러그인을 폭넓게 개발할수 있을 뿐만아니라 제공되고 있다. fluentd는 다른 fluentd에 전달도 가능한데, 이를 통해 fluentd의 트래픽을 조정하거나 라우팅할 수도 있다. 아키텍쳐 단순성과 안정성으로 인해 많은 IT기업들에서 사용된다. 파이프라인 아키텍쳐 구상 및 준비 앞서 말했듯이 강력한 input/output 플러그인 기능을 가지고 있는데, 실제로 어떤 configuration으로 사용 가능할지 알아보기 위해 아래와 같은 아키텍쳐를 구현해보기로 하였다. 상기 아키텍쳐에서 파이프라인은 2개로 나뉘어져 있다. 1)..

Telegraf와 fluentd는 아주 유사해보인다. 둘다 configuration파일 기반으로 작동하며 plugin을 통해 개발자가 custom하게 만든 input, filter, output 플러그인들을 사용하여 데이터를 처리, 전송 가능하다. 이 둘은 어떤 차이가 있는 것일까? Fluentd Star count(9/16) : 8,295 Github url : https://github.com/fluent/fluentd Fluentd는 C와 루비로 작성된 로그 수집기(log aggregator)의 한 종류이다. 여러 데이터 소스에서 데이터를 수집해 오는 부분에 있어 컴포넌트로서의 역할을 한다. fluentd의 아키텍쳐 단순성과 안정성에 초점을 두고 사용된다. Fluentd의 내부구조 - Input :..

Kafka의 consumer를 사용할 때 offset에 대한 commit offset timing에 대해 여러가지 방법이 있다. 만약 enable.auto.commit=true 로 사용시 consumer에서 close()를 호출한다면 어떻게 kafka 내부에서 offset을 처리하는지 확인해보려고 한다. (아래는 kafka consumer 2.1.0 기준으로 작성되었습니다) 1. close() 호출 @Override public void close() { close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS)); } Kafka의 consumer가 close될 때 timeout 시간 내에 consumer를 종료한다.(default는 30초) 만약 auto-commit=tr..

mongodb shell에서 print()는 javascript의 method로서 사용가능하여 bson과 같은 데이터를 text로 표현하지 못한다. 그러므로 bson을 데이터로 보기 위해서는 mognodb에서 제공하는 printjson() method를 사용하면 된다. Example 1 > var dblist=["server01","server02"]; > dblist.forEach(function(e){ var serviceCollectionData = db.getSiblingDB(e).service.findOne(); print(serviceCollectionData); // print method 사용 }); [object BSON] [object BSON] 기본 print method를 사용하면 ..

이전 포스팅(mongoDB 개요 및 설명 한페이지에 끝내기)에서 언급한것 처럼 mongodb내에 js script를 적용할 수 있다. 이를 활용해서 여러 db의 값을 동시에 조회할 수 있다 요구사항 DB list : server01, server02, server03 Collection : service -> 각 DB의 service collection에 있는 데이터를 모두 조회하라 Code > var dblist=["server01","server02","server03"]; > dblist.forEach(function(e){ var serviceCollectionData = db.getSiblingDB(e).service.findOne(); printjson(serviceCollectionData)..

confluent에서 kafka에 관련된 지식 퀴즈 14개를 아래와 같이 public에 공개하였다. 간단하지만 핵심적인 kafka 질문들을 모아놓아서 재밌어서 각 질문들에 대해 정리해보고자 한다. quiz site : https://www.surveymonkey.com/r/FundamentalsSelfAssess 각 문제에 대한 정답은 정답 : 뒤를 드래그 하면 보인다. 1. Kafka에서 record가 실리는 곳은? (1) Category (2) Subject (3) Topic (4) Table 정답 : (3) Topic 2. Kafka에서 각 데이터(record)들이 파티션 내부에서 유니크하게 구분되는 값은? (1) Primary Key (2) Offset (3) Identifier (4) Times..

이전 블로그포스팅에서 getSiblingDB라는 db object를 반환하는 method에 대해서 알아보았다. mongodb에서 db 이름 명시하여 데이터 조회하기 Mongodb를 사용하다보면 use [dbname]을 써서 데이터를 collection단위로 조회하는 경우도 있지만 db이름을 명시하여 데이터조회하고 싶을 때가 있다. 그때는 db.getSibligDB() method를 사용하면 된다. db.getSi.. blog.voidmainvoid.net 상기 method를 사용해서 서로 다른 database의 데이터를 비교하는 구문을 만들어보자 요구사항 - students database와 teacher database가 존재 - students에는 list라는 collection이 존재하고 stude..

Mongodb를 사용하다보면 use [dbname]을 써서 데이터를 collection단위로 조회하는 경우도 있지만 db이름을 명시하여 데이터조회하고 싶을 때가 있다. 그때는 db.getSibligDB() method를 사용하면 된다. db.getSiblingDB() parameter - : 몽고db의 데이터베이스 이름 return - Database object Example getSiblingDB method를 사용해서 아래와 같이 database object를 가지고 유연한 database 조회 및 사용이 가능하다. teachers = db.getSiblingDB('teachers') // teachers db return students = db.getSiblingDB('students') // ..

개요 Elasticsearch과 kibana를 통해 document를 저장하고 시각화 할 수 있다. Elasticsearch에 document를 넣기 위해 여러 언어를 사용할 수 있지만 python을 사용하면 한결 간단한 코드로 작성이 가능하다. Elasticsearch Client List는 아래와 같다. - Java REST Client [7.3] - Java API [7.3] - JavaScript API [7.x] - Ruby API [7.x] - Go API .NET API [7.x] - PHP API [7.0] - Perl API - Python API - Community Contributed Clients 출처 > https://www.elastic.co/guide/en/elasticsea..

이전 포스팅에서 Kafka burrow의 정보와 lag을 평가하는 방법에 대해 알아보았다. Burrow 개요 및 설명 : https://blog.voidmainvoid.net/243 Burrow에서 lag을 정의(평가)하는 방법 : https://blog.voidmainvoid.net/244 이번 포스팅에서는 Burrow에서 제공하는 http endpoint를 알아보고 각 topic별 offset, lag등의 정보를 가져오는 예제를 알아보고자 한다. Burrow Endpoint Burrow의 http server는 kafka, zookeeper의 정보를 효과적으로 가져올 수 있다. 모든 요청은 간단한 HTTP call로 요청되며(대부분 GET request) response는 JSON기반이다. Healt..

이전 포스팅(https://blog.voidmainvoid.net/243)에서 Burrow가 나오게된 배경에 대해 알아보았다. 이 포스팅에서는 burrow가 lag의 상태에 따라 상태를 정의하는 방법에 대해 알아보자. Consumer Lag Evaluation Rules Burrow에 있는 consumer group의 상태는 group이 consume하고 있는 각 partitin에 대한 offset의 규칙에 따라 결정된다. 분리된 threshold를 정하지 않더라도 이 kafka consumer들이 '정상'적으로 작동중인지, '비정상'적으로 작동중인지 판단 할 수 있다. consumer group이 consume하는 모든 파티션에 대해 평가를 함으로서 consumer group이 정상적으로 consume..

Burrow github : https://github.com/linkedin/Burrow Burrow는 Kafka를 개발한 Linkedin에서 만든 consumer lag monitoring tool이며, opensource로 운영되고 있다. 각 consumer는 특정 topic에 대해 고유의 groupId를 가지고 consuming을 하는데, consume이 잘 되고 있는지 모니터링이 필요하다. Burrow가 나오게된 배경 기존 Kafka client의 consumer의 metrics() method를 사용하여 lag metric(records-lag-max)을 기록할수 있지만, 이는 가장 뒤처진 파티션의 현재 지연을 보여주므로 다른 파티션에서의 정상작동을 잘 감지하기가 어렵다. 또한, consume..

MongoDB 주요 특징 Secondary Index ▪ 다른 NOSQL 보다 secondary index 기능이 발달되어 있음 샤드키 지정 ▪ _id : 키 필드 ▪ Shard Key _id - 대부분의 NOSQL은 Row Key = Shard Key 임 Document 기반 ▪ JSON Style의 Document : BSON(Binary JSON) Modelling 주요 적용 모델링 기법 ▪ 비정규화(Denormalization) ▪ 집합(Aggregation) 기타 mongoDB에 대한 정보 → https://blog.voidmainvoid.net/239 NoSQL강의) mongoDB 개요 및 설명 한페이지에 끝내기(mapReduce, aggregate 예제 포함) Humongous DB ▪ Do..

모델링 example 1 - Log 데이터 저장요구조건 - machine_id , log_time, log_text - machine 100EA, 1초당 로그 1건씩, 로그는 일반 텍스트 Case 1 : 별도의 PK를 부여한다면? CREATE TABLE log1 ( uid int, machine_id varchar, log_time timestamp, log_text varchar, PRIMARY KEY (uid) ); → 특정머신의 로그만 조회하는 것이 불가. 의미없는 모델. ▪ Log성 테이블은 Primary Key를 단일 값으로 설정하는 것이 힘들다. machine_id를 단일 primary key로 설정하면 machine_id가 Row Key가 되는데, 이 경우 시계열 데이터를 저장할 수 없게 된..