데브원영

카프카 스트림즈의 KTable은 토픽의 데이터를 key-value형태로 사용할 수 있도록 구체화된 뷰(Materialized View)를 제공합니다. 구현방법은 다음과 같습니다. 0. 카프카 스트림즈 디펜던시 추가 dependencies { testImplementation 'org.junit.jupiter:junit-jupiter-api:5.7.0' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.7.0' compile 'org.slf4j:slf4j-simple:1.7.30' implementation "org.apache.kafka:kafka-streams:2.8.0" } 이 예제는 2.8.0을 기준으로 작성하였습니다. 1. 카프카 스트림즈 설정..

Tranquility는 아파치 드루이드에 실시간으로 데이터를 적재하기 위한 용도로 사용됩니다. 아파치 드루이드에 대한 정보는 아래 링크를 사용해주세요. https://voidmainvoid.tistory.com/440 아파치 드루이드 소개 및 아키텍처 Apache Druid is a high performance real-time analytics database. 아파치 드루이드 소개 아파치 드루이드는 기존의 데이터 처리 및 쿼리에 대한 관념을 180도 바꿔주는 OLAP 데이터베이스 입니다. OLA.. blog.voidmainvoid.net 현재 버전(0.20.0)의 드루이드는 카프카와 같은 데이터 소스로 부터 데이터를 직접 적재(ingestion)할 수 있지만 이전 버전(0.8.0)에는 그렇지 않았습..

Apache Druid is a high performance real-time analytics database. 아파치 드루이드 소개 아파치 드루이드는 기존의 데이터 처리 및 쿼리에 대한 관념을 180도 바꿔주는 OLAP 데이터베이스 입니다. OLAP란 Online Analytical Processing의 약자로써 사용자가 적재한 데이터를 다양한 방식(다차원)으로 적재하고 분석하도록 도와주는 시스템을 뜻합니다. 다차원 정보라는 것은 기존에 1차원 정보(row단위)를 몇개의 필드들(드루이드에서는 디멘젼)을 사용하여 지표들(메트릭)으로 만들어 보여주는 것입니다. 흥미롭게도 아파치 드루이드는 이러한 다차원 데이터를 빠른 쿼리속도를 제공하기 위해 세그먼트 단위로 데이터를 인덱스하여 저장합니다. 이렇게 세그먼..

카프카와 연동하는 스파크 스트림은 2가지 방식으로 구현할 수 있습니다. 1. 구조적 스트림(DataFrame readStream) 2. 스파크 스트리밍(DStream) 그 중 구조적 스트림 방식을 gradle + scala로 구현하기 위한 방법을 설명합니다. build.gradle plugins { id 'idea' id 'java' id 'scala' } group 'com.example' version '1.0-SNAPSHOT' sourceCompatibility = 1.8 targetCompatibility = 1.8 repositories { mavenCentral() jcenter() } ext { scalaVersion = '2.12.14' sparkVersion = '3.1.2' } depe..

카프카 스트림즈는 기본적으로 at-least-once 프로세싱을 지원합니다. at-least-once 프로세싱이란 적어도 한번 프로세싱을 지원하는 것으로써 중복은 발생할 수 있고 유실은 절대 발생하지 않는 것입니다. 스트림즈 프로세싱에서라면 특정 레코드에 대해 처리를 하고 유실은 될 수 없고 중복 프로듀스는 될 수 있다는 것입니다. - at least once : 적어도 한번 처리 - at most once : 많아도 한번만 처리 - exactly once : 딱 한번만 처리 카프카는 0.11.0 이후 버전에서 transaction처리를 지원합니다. 이 옵션을 사용하면 스트림즈 API를 사용할 때 Exactly once 프로세싱을 만족하도록 적용할 수 있습니다. 적용하는 방법은 아래와 같습니다. 코드를 ..

카프카 스트림즈 사용시 토폴로지 내부 에러로 인해 아래와 같은 오류가 발생할 수 있다. All stream threads have died. The instance will be in error state and should be closed. 이에 대응하기 위해 2.8.0에서 추가된 setUncaughtExceptionHandler메서드를 사용하면 스레드를 재시작하여 데이터를 지속 처리할 수 있다. streams.setUncaughtExceptionHandler((exception) -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD); 2.8.0 이전에는 setUncaughtExceptionHandler(fin..

mongodb 자바 라이브러리 정의(in gradle) dependencies { implementation "org.mongodb:mongo-java-driver:3.4.1" } mongoClient 선언 MongoClient mongoClient = new MongoClient("localhost", 27027); MongoCollection collection = database.getCollection(COLLECTION_NAME); upsert 수행 String findKey = "{\"name\":\"wonyoung\"}"; // 찾아야할 데이터 값 int count = 1; // 추가하는 데이터 BasicDBObject keyObject = BasicDBObject.parse(findKey)..

무엇이 추가되었을까요? - GPU 렌더링 - 애플 실리콘(M1 시리즈), arm 지원 - 탭 멀티셀렉트 지원 - 오토 컴플릿트 지원(context-aware) - UI개선 - 타입스크립트 지원 - syntax 정의 지원 - 파이썬 api 개선 관련 내용 : https://www.sublimetext.com/blog/articles/sublime-text-4 Sublime Text 4 Meet the new Sublime Text - it's faster and smarter than ever with hardware acceleration, Apple silicon support, and more! www.sublimetext.com 다운로드 : https://www.sublimetext.com/ Sub..

실행 가능한 자바의 jar 패키지를 만들기 위해 shallowjar 등을 사용했었다. 그러나 아래와 같은 코드를 사용하여 uberJar(디펜던시가 모두 포함된)을 만들 수도 있다. task uberJar(type: Jar) { from sourceSets.main.output dependsOn configurations.runtimeClasspath from { configurations.runtimeClasspath.findAll { it.name.endsWith('jar') }.collect { zipTree(it) } } manifest { attributes "Main-Class": "com.example.Main" } } - package명 : com.example - public static ..

Unixtime으로 현재시간을 가져오고 싶다면 아래와 같이 수행합니다. System.currentTimeMillis() Returns the current time in milliseconds. Note that while the unit of time of the return value is a millisecond, the granularity of the value depends on the underlying operating system and may be larger. For example, many operating systems measure time in units of tens of milliseconds. See the description of the class Date for a ..

테스트를 위해 랜덤 숫자를 추출해야할 때가 있습니다. 이때는 Math함수를 사용하면 됩니다. 0부터 9까지 랜덤 숫자 (int)(Math.random()*10) 0부터 99까지 랜덤 숫자 (int)(Math.random()*100) 1부터 100까지 랜덤 숫자 (int)(Math.random()*100) +1

ProducerRecord는 다양한 파라미터를 받고 그 중 파티션번호를 직접 넣어서 사용하는 경우도 있습니다. 이 때 파티션 번호를 넣으면 어떻게 동작할까요? 정답은 KafkaProducer private partition()메서드를 보면 확인할 수 있습니다. /** * computes partition for given record. * if the record has partition returns the value otherwise * calls configured partitioner class to compute the partition. */ private int partition(ProducerRecord record, byte[] serializedKey, byte[] serializedV..

1. regex를 통해 프로메테우스 데이터 모델의 레이블 값 추출하기 프로메테우스의 메트릭은 아래와 같은 형태로 되어 있습니다. {=, = ...} [] 그라파나의 variable에 레이블 값을 사용하고 싶을 때는 아래와 같이 설정합니다. 1) Variable 생성, 설정 2) Query Options 설정 Data source : 메트릭이 존재하는 프로메테우스 선택 Query : 메트릭 이름 입력(여기서는 burrow 수집 데이터) Regex : 정규식 형태로 레이블 키에 속한 레이블 값 뽑아내기 /.*group="([^"]*).*/ 위와 같은 정규식 형태를 사용해서 prefix를 가지고 있는 레이블 값만 뽑아낼 수도 있습니다. /.*group="((spark-)[^"]*).*/ spark- 로 시작하..

우선 이벤트 소싱과 CQRS에 대한 개념부터 짚고 넘어가겠습니다. 카프카에 대한 개념은 이 동영상을 참고해주세요. 이벤트 소싱 패턴 이벤트 소싱은 배치성 데이터 저장소에 현재 상태만 저장하는 것이 아니라 이벤트 브로커(전용 저장소)에 발생된 모든 이벤트 기록(레코드)들을 기록하는 것을 뜻합니다. 이를 통해 데이터 모델과 비즈니스 도메인을 동기화할 필요가 없어지고 성능, 확장성, 응답성이 향상되어 도메인 태스크를 간소화 할수 있습니다. 기존에는 배치성 데이터 저장소(오라클, mysql 등)에 서비스에 들어오는 명령(또는 이벤트)이 발생할 때 마다 항상 동기화 했었습니다. 간단한 구조와 적은 데이터 접근만 일어날 때는 큰 이슈가 없지만, 데이터과 이벤트가 많아질 수록 데이터에 대한 CRUD와 같은 접근으로 ..

confluent-kafka-go는 컨플루언트에서 개발하고 유지보수하는 golang기반 카프카 클라이언트입니다. github.com/confluentinc/confluent-kafka-go confluentinc/confluent-kafka-go Confluent's Apache Kafka Golang client. Contribute to confluentinc/confluent-kafka-go development by creating an account on GitHub. github.com 위 라이브러리를 사용해서 컨슈머 5가지 패턴을 구현해보겠습니다. 컨슈머 구현 사전 작업 코드(컨슈머 초기화) 이 작업은 컨슈머를 이용하기 위한 가장 기본적인 작업입니다. 기본 옵션은 enable.auto.com..

레디스란? 레디스는 오픈소스 인 메모리 데이터 구조 저장소로서 데이터베이스, 캐시, 메시지 브로커로 역할을 수행한다. 레디스는 String, hash, lists, sets, sorted sets, bitmaps, streams등을 지원한다. 레디스 큐란? 레디스 큐는 레디스의 자료구조 중 List를 이용하여 Queue를 구현한 것이다. 큐는 FIFO(First In First Out)구조로 먼저 들어온 데이터가 먼저 처리되는 것이다. LIST자료구조의 LPUSH, RPOP(또는 RPUSH, LPOP)을 사용하여 구현할 수 있다. 이외에도 BLPOP을 사용하여 블락킹(blocking) pop을 수행할 수 있다. 지정한 시간만큼 기다리고 값이 들어오면 LPOP을 수행하는 것이다. $ lpush my-que..

아래 리포지토리를 참고하여 생성하였습니다. github.com/container-examples/alpine-telegraf container-examples/alpine-telegraf Docker image Telegraf (Alpine). Contribute to container-examples/alpine-telegraf development by creating an account on GitHub. github.com FROM alpine:3.7 ENV TELEGRAF_VERSION 1.18.1 WORKDIR /scripts ADD https://dl.influxdata.com/telegraf/releases/telegraf-${TELEGRAF_VERSION}_static_linux_amd6..

Metric buffer overflow; 3645 metrics have been dropped 상기와 같이 오류가 나는 이유는 telegraf agent 내부에서 사용하는 버퍼가 모두 찼기 때문입니다. 아래와 같이 buffer를 늘려서 대응할 수 있습니다. [agent] metric_buffer_limit = 10000 관련정보 metric_buffer_limit: Maximum number of unwritten metrics per output. Increasing this value allows for longer periods of output downtime without dropping metrics at the cost of higher maximum memory usage. github..

1. 다운로드 링크 : spark.apache.org/downloads Downloads | Apache Spark Download Apache Spark™ Choose a Spark release: Choose a package type: Download Spark: Verify this release using the and project release KEYS. Note that, Spark 2.x is pre-built with Scala 2.11 except version 2.4.2, which is pre-built with Scala 2.12. Spar spark.apache.org 2. 압축풀기 저는 Documents 아래에 압축을 풀었습니다. $ pwd ~/Documents/spark-3..

바이오필리아 가설(Biophilia hyphothesis)란? 인간은 자연과의 접점을 유지하려는 본능을 가지고 있다는 뜻. 그리고 이러한 욕구가 충족될 때 일터에서도 더 큰 활력과 의욕을 가지게 된다. 회사에 자연의 요소를 녹여 놓은 바이오필릭(biophilic, 자연 친화적) 사무실을 설계하고 원격 근무자도 외출을 장려해야한다고 강조한다. 추가적으로 가상 회의실 배경을 자연으로 꾸미는 등 세심한 배려를 전략적으로 시도할 필요가 있다고 강조했다. 일터는 어떻게 바뀌어야할까? 캘리포니아 쿠퍼티노의 애플 신사옥은 바이오필릭 원칙을 수용하고 있다. 인공 구조물과 녹지를 자연스럽게 연결해 녹여냈다. 그리고 시애틀에 위치한 아마존의 스피어 본사는 유리로된 원형 건물내에 수천종의 식물을 심었다. 내부에는 현수교 정..

macOS에 minikube 설치 $ brew install minikube kubectl 설치 $ brew install kubectl minikube 실행 $ minikube start 대시보드 확인 $ minikube dashboard minikube 일시중지 $ minikube pause minikube 중단 $ minikube stop minikube 삭제 $ minikube delete --all 링크 : minikube.sigs.k8s.io/docs/start/

데브원영 유튜브 채널을 운영하면서 촬영, 편집을 하고 있는데요. 최근에 찍은 영상을 기준으로 영상 장비와 셋팅값을 정리합니다. 장비 종류 - SONY icd-tx650 보이스 레코더 - panasonic gx9 미러리스 카메라 - panasonic 20mm f1.7 렌즈 - 선웨이포토 FL-152 x2 장비 셋팅 1) 보이스 레코더 - 녹음모드 : MP3 192kbps - 녹음환경 선택 : 끔 - 마이크감도 : 낮음 - 스테레오 강조 : 끔 - 녹음필터 : 끔 - VOR : 끔 2) gx9 미러리스 카메라 - MF 셋팅. Auto focus 사용시 촛점 울렁거림으로 별로임. - CNEV(영화같은 비디오) - MP4, FHD, 30p - 수동 측광모드, 상황에 따라 조리개, ISO, 셔터 스피드 조절하여 ..

카프카 스트림즈에서 commit.interval.ms 옵션은 프로세서의 frequency 즉, 반복을 얼마만큼 인터벌 간격으로 할 것인가에 대한 것이다. The frequency in milliseconds with which to save the position of the processor. (Note, if processing.guarantee is set to exactly_once, the default value is 100, otherwise the default value is 30000. 예를 들어 KTable로 선언된 변수를 toStream으로 변경하는 구문이 있다고 가정하자. 이 때 30000이면 30초 인터벌을 가지고 변경을 수행한다. 만약 짧게 가져가고 싶다면 0이나 아주 작은 숫..

2020년 6월 부터 집필을 시작하여 6개월간의 원고 집필과 3개월간의 편집 끝에! 「아파치 카프카 애플리케이션 프로그래밍 with 자바」를 출간하게 되었습니다. 출간일은 4월 14일로 화요일부터 구매 순서대로 배송이 시작됩니다. - 예스24 : www.yes24.com/Product/Goods/99122... - 교보문고 : www.kyobobook.co.kr/product/detai... - 알라딘 : www.aladin.co.kr/shop/wproduct.aspx?... 책을 집필하게 된 계기 사실 개발 블로그와 개발 유튜브를 진행하면서 딱히 책을 쓸 생각은 없었습니다. 책을 쓰는데 들이는 시간과 노력이 생각보다 크다는 것을 알고 있었기 때문입니다. 하지만 제 컨텐츠들을 봐주시고 '카프카를 주제로'..

카프카 스트림즈에는 suppress()메서드가 있다. 이 메서드를 통해 일정 윈도우 간격 안에 데이터를 모아서 처리할 수 있는데, 이슈가 있다. 값을 모아서 state store(rocksDB)에 저장할 때 제대로 deserialize를 하지못하는것이다. java.lang.ClassCastException: org.apache.kafka.streams.kstream.Windowed cannot be cast to java.lang.String at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28) at org.apache.kafka.streams.kstream.internals.suppress..

이 에러는 카프카 스트림즈 내부에서 null 메시지 값에 대한 처리를 제대로 못해서 나오는 이슈이다. 그러므로 KStream에서 filter로 null 메시지 값을 삭제하여 처리하면 된다. KTable KT01 = myStream .filter((k, v) -> v != null) // null value 제거 ...

카프카 스트림즈 애플리케이션을 운영하다가 또는 테스트 중에 오프셋을 초기화 해야할 때가 있습니다. 이때는 아래와 같은 명령어를 사용하면 됩니다. $ ./kafka-streams-application-reset.sh --application-id join-application --input-topics streams-test Reset-offsets for input topics [A.s2olap-kakaotalk] Following input topics offsets will be reset to (for consumer group join-application) Topic: streams-test Partition: 2 Offset: 0 Topic: streams-test Partition: 3 Of..

2020년은 '교육'이 미래에 어떤 방향으로 바뀔지 방향을 알려주는 해였다고 생각됩니다. 코로나의 직접적인 감염으로 인해서 더 이상 오프라인으로 사람과 사람이 만날 수 없었고 학교와 기업에서 진행되어야 하는 교육이 온라인으로 대체 되었습니다. 온라인 교육은 어떻게 수행되어야하고 어떤 미래로 가야하는지 정리해보았습니다. '형식적인' 교육은 그만 기존의 많은 기업들은 교육을 진행할때 '형식'에 집중하였습니다. 최근까지 많은 기업 교육은 SCORM(Sharable Content Object Reusability)이라고 불리는 미국이 만든 웹 기반 교육 표준에 따라 많은 교육 컨텐츠들이 만들어져 있었습니다. 기업에서 윤리, 부패방지 등과 같은 교육을 들으신 분은 아실 테지만 화면 아래에 진도가 표시되는 막대기가..

DSP(demand side platforms) 광고 구매 플랫폼. RTB(real time bidding) 실시간 경매 시스템 SSP(supply side platforms) 공급자측 플랫폼 광고주 광고를 지면에 노출시키고 싶은 사업자 또는 개인 매체(또는 인벤토리) 광고를 노출시킬 수 있는 지면을 가지고 있는 플랫폼 CPM(cost per mile impression) 광고 노출 1천회를 기준 요금 부과 방식 CTR(click through rate) 100명 광고 노출당 클릭 횟수 지표 CVR(conversion rate) 광고 클릭 대비 전환율. 클릭 이후에 다운로드/구매 등의 행동이 일어났는지 지표 eCPM(effective CPM) CPM, CTR, CVR을 하나의 지표로 환산한것 ROAS(r..

golang은 동시성 프로그램을 위해 태어난 언어라고 해도 과언이 아닙니다. 예제를 통해 golang이 동시성을 만족하는 코드를 작성하여 보여드리겠습니다. 동시성을 위해 goroutine과 channel 개념을 사용합니다. 각 개념은 아래 링크에서 확인할 수 있습니다. - goroutine : golang.site/go/article/21-Go-%EB%A3%A8%ED%8B%B4-goroutine Go루틴(goroutine)은 Go 런타임이 관리하는 Lightweight 논리적 (혹은 가상적) 쓰레드(주1)이다. Go에서 "go" 키워드를 사용하여 함수를 호출하면, 런타임시 새로운 goroutine을 실행한다. goroutine은 비동기적으로(asynchronously) 함수루틴을 실행하므로, 여러 코드를..