데브원영

원문 : https://blogs.nvidia.com/blog/2020/09/03/what-is-mlops/ What is MLOps? Machine learning operations are best practices for businesses to run AI successfully. blogs.nvidia.com MLOps란 무엇인가? MLOps(Machine learning operations)는 AI를 통해 비즈니스 발전을 도와주는 소프트웨어 제품들과 클라우드 서비스입니다. MLOps는 익숙하지 않지 않은 단어이지만, 기업의 AI를 성공시키기 위한 두 단어로 이루어진 용어이다. Machine learning operation을 줄인 용어인 MLOps는 비즈니스에서 AI를 성공적으로 적용시키기 위..

원문 : https://netflixtechblog.com/exploring-data-netflix-9d87e20072e3 Exploring Data @ Netflix By Gim Mahasintunan on behalf of Data Platform Engineering. netflixtechblog.com 어느 조직이든 빠르게 성장하는 가운데 다양한 데이터를 어떻게 저장하고 사용할지 결정하는 것은 매우 어려운 일이다. 넷플릭스에서는 스트림 데이터를 쉽게 사용할 수 있는 툴을 개발하고 활용하고 있다. 이번 포스트를 통해 Netflix Data Explorer 툴을 소개하게 되어 매우 기쁘다. Data Explorer는 엔지니어들에게 카산드라, 다이너마이트, 레디스 저장소에 있는 데이터를 빠르고 안전하게..

카프카 스트림즈는 비상태기반(stateless), 상태기반(stateful) 처리를 지원하는 다양한 메서드를 제공합니다. 스트림즈DSL을 통해 구현할 수도 있고 또는 프로세서API를 사용해서 직접 구현하는 방식도 있습니다. 이 포스팅에서는 스트림즈DSL을 활용하여 상태기반 데이터 처리할 때 어떻게 input, output이 동작하는지 설명합니다. 카프카 스트림즈의 스트림즈DSL에서 window function은 4가지가 있습니다. - 텀블링 윈도우 - 세션 윈도우 - 슬라이딩 윈도우 - 호핑 윈도우 이 중 가장 텀블링 윈도우에 대해 살펴볼건데요. 텀블링 윈도우는 특정 사이즈의 윈도우가 서로 다른 윈도우와 겹치지 않게 지속되는 시간 단위 윈도우를 뜻합니다. 다음은 텀블링 윈도우 예시 사진입니다. 레코드의 ..

원문 : https://kafka-tutorials.confluent.io/kafka-streams-schedule-operations/kstreams.html 카프카 스트림즈는 토픽의 데이터를 읽어 상태기반, 비상태기반 처리를 하는 스트리밍 라이브러리입니다. 오늘은 컨플루언트에서 카프카 스트림즈가 스케쥴링 동작을 어떻게 수행하는지에 대한 코드 예시를 번역하였습니다. 1. 프로젝트 초기화 신규 디렉토리를 생성합니다. $ mkdir kafka-streams-schedule-operations && cd kafka-streams-schedule-operations 2. 컨플루언트 플랫폼 가져오기 Dockerfile을 통해 데이터 생성기를 먼저 가져옵니다. 하기 Dockerfile은 Dockerfile-con..

프로메테우스로 다음과 같은 지표들을 쌓을 때가 있습니다. http_request_count_total{method="POST",router="/"} 10 http_request_count_total{method="POST",router="/"} 15 http_request_count_total{method="POST",router="/"} 20 http_request_count_total{method="POST",router="/"} 40 http_request_count_total{method="POST",router="/"} 45 http_request_count_total{method="POST",router="/"} 60 상기와 같은 지표는 http request때 마다 counter를 1씩 in..

프로메테우스 promQL을 사용할 경우 Aggregation Operators와 함께 사용하는 경우가 빈번하다. - sum (calculate sum over dimensions) - min (select minimum over dimensions) - max (select maximum over dimensions) - avg (calculate the average over dimensions) - stddev (calculate population standard deviation over dimensions) - stdvar (calculate population standard variance over dimensions) - count (count number of elements in the..

프로메테우스는 지표를 모니터링하기에 적합합니다. 지표를 모니터링하면 애플리케이션 동작에 대한 상태변화를 즉각적으로 알고 대처할 수 있습니다. 자바애플리케이션에서 프로메테우스로 지표를 보내는 방법은 크게 3가지 방법이 있습니다. - JMX 익스포터 - PUSH 게이트웨이 - 자바 클라이언트 여기서는 자바 클라이언트를 사용하는 방법을 설명합니다. https://github.com/prometheus/client_java prometheus/client_java Prometheus instrumentation library for JVM applications - prometheus/client_java github.com 우선 build.gradle에 관련 라이브러리를 추가합니다. dependencies {..

카프카 스트림즈는 상태기반/비상태기반 데이터 처리에 효과적인 라이브러리입니다. 특히 상태 기반 처리에 큰 도움이 되는데 여러 상태 기반 처리 중 SlidingWindow에 대해 살펴보고자 합니다. SlidingWindow는 window종류 중 하나로서 일정 시간동안의 데이터들의 집합에 대해 연산을 하는 것을 뜻합니다. 마치 베란다의 슬라이딩 윈도우가 옆으로 지나가는 듯한 모습과 비슷합니다. 카프카 스트림즈에서는 aggregation연산 또는 join연산을 할 때 window를 적용할 수 있는데 총 4개의 윈도우를 지원합니다. 각 윈도우 이름과 특징은 다음과 같습니다. - 호핑 윈도우 : 고정적인 사이즈의 윈도우, 윈도우끼리 겹치는 부분이 있음 - 텀블링 윈도우 : 고정적인 사이즈의 윈도우, 윈도우끼리 겹..

옛날버전(0.9이하)에서 외부 실시간 데이터를 드루이드에 넣기 위해서는 tranquility를 사용해야 했습니다. tranquility-core를 사용하여 데이터를 넣을 때 아래와 같은 오류를 경험할 때가 있습니다. com.metamx.tranquility.tranquilizer.MessageDroppedException: Message dropped MessageDroppedException는 줄여서 MDE라고도 부르는데요. 이 오류가 발생하게 되는 원인은 크게 두가지라고합니다. 원인1 : 이벤트의 timestamp가 window period 외부에 있을 경우 원인2 : 태스크 또는 오버로드와 통신에 이슈가 있는 경우(within your firehoseRetryPeriod or indexRetryPe..

카프카 스트림즈에서 KStream 또는 KTable을 사용하여 join을 사용할 때가 있습니다. KStream completedEventsStream = leftStream. join( rightStream, (leftValue, rightValue) -> leftValue + rightValue, JoinWindows.of(windowDuration) ); 상기와 같이 leftStream과 rightStream 2개의 스트림데이터를 조인하는 것은 아주 일반적인 조인 사용 예시인데요. 여기서 lambda 식을 보면 알 수 있다 싶이 leftValue와 rightValue에만 접근이 가능합니다. 즉, 2개의 토픽에서 조인이 되는 조인 key에 대해서는 접근이 불가능하다는 것을 알 수 있습니다. 2개의 토픽..

KTable을 Materialized View로 사용할 경우 아래와 같은 에러가 발생할 때가 있습니다. Exception in thread "Timer-0" org.apache.kafka.streams.errors.InvalidStateStoreException: Cannot get state store address because the stream thread is STARTING, not RUNNING at org.apache.kafka.streams.state.internals.StreamThreadStateStoreProvider.stores(StreamThreadStateStoreProvider.java:81) at org.apache.kafka.streams.state.internals.Wr..

카프카 스트림즈의 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)..

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..

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..

카프카 스트림즈에서 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..