데브원영

이 포스트는 2020년 2월 18일 기준입니다. Scala는 JVM위에서 돌아가는 언어이며, Scala버젼별로 JVM 버젼의 호환성이 있으므로 아래 표를 기준으로 Scala버젼을 운영해야합니다. Scala 공식홈페이지에 따르면, Java 8 이상에서 scala를 compile하는 것을 권장합니다. JVM은 하위호환성이 있으므로, Compile은 옛날버젼으로 하되, 실제 code를 Running할때는 최신버젼의 JVM버젼을 사용하는게 일반적입니다. Java 8으로 compile된 Scala code는 JVM 8 이상에서 정상적으로 돌아야만 합니다. 만약 Java8으로 compile했음에도 불구하고 이슈가 생긴다면 이는 긴급수정 사항입니다. 일부 오래된 JVM버젼은 scala code의 compile과 ru..

pyspark를 활용할때 dataframe의 변환과 같은 작업을 위해 UDF가 필요할 때가 있습니다. UDF란 User Defined Functions의 약자로서 사용자가 직접 개발한 method를 뜻합니다. UDF를 만들기 위한 sample 코드를 공유하고자 합니다. 그리고 pyspark에서 dataframe을 처리하려면 아래와 같이 사용하곤합니다. df = sc.read.csv(target_data,header=False,sep="\u0001") processedDf = df.withColumn(target_column, customMethod(df[target_column])) 그리고 Spark에 UDF를 등록하기 위해 아래와 같이 annotation을 붙여주는 방식으로 사용하여 function을 ..

pyspark를 통해 다양한 파일을 읽을 수 있습니다. 보통 .text 또는 .log와 같은 확장자로 되어진 plainText를 읽기도 하지만 압축된 파일인 .gz 과 같은 파일을 읽어야할 때도 있습니다. 이렇게 .gz과 같이 압축된 파일을 pyspark를 통해 읽으려면 어떻게 해야할까요? rdd = sc.textFile("data/label.gz") print rdd.take(10) 정답은 생각보다 간단합니다. textFile method를 사용하여 .text나 .log파일을 읽듯이 그대로 입력하여 읽으면 됩니다. Spark Document에 따르면 아래와 같이 나와 있습니다. All of Spark’s file-based input methods, including textFile, support r..

pyspark로 개발하다 보면 local에서 돌릴때는 정상적으로 돌아가는데 cluster mode(yarn 등)으로 실행할때 아래와 같이 오류가 발생하는 경우가 있습니다. 사실 spark가 돌아가는 방식에 대해 이해하면 쉬운데, 결론부터 말하자면 worker node에 해당 모듈이 깔려 있지 않아서 발생하는 오류입니다. 위 그은Spark가 Cluster deployment mode로 돌아가는 것을 형상화 합니다. 실제로 돌아가는 Task는 각 worker node에서 수행하기 때문에 각 worker node에 module 이 존재하지 않으면 module을 못찾게 되는 것입니다.

Python으로도 암호화를 손쉽게 할 수 있습니다. 이번 포스팅에서는 AES/CBC/PKS5Padding으로 암호화 하는 Python code를 보여드리겠습니다. 먼저 필요한 module은 아래와 같습니다. from Crypto.Cipher import AES 위 모듈을 사용하기 위해서는 아래와 같이 pip를 통해 모듈을 설치해줍니다. $ pip install pycrypto 모듈 사이트 : https://pypi.org/project/pycrypto/ pycrypto Cryptographic modules for Python. pypi.org AES encrypt를 하기 위해서는 key와 iv가 필요하므로 선언하며, 암호화 길이가 32보다 작을경우 padding으로 빈공간을 채워주는 로직을 사용합니다..

카프카에서 consumer를 새로 생성하여 topic에서부터 데이터를 가져오기 위해서는 여러 옵션이 필요한데 그 중 하나는 auto.offset.reset입니다다. 이 auto.offset.reset의 역할에 대해 알아보겠습니다. 아래 글은 Kafka-client, Kafka broker 2.4 기준으로 작성하였습니다. auto.offset.reset auto.offset.reset에서 offset은 consumer offset입니다. 만약 이번에 topic에 붙은 consumer의 offset정보가 존재하지 않는다면 auto.offset.reset의 default값(latest)이나 또는 설정한 값을 따라가게 됩니다. auto.offset.reset - latest : 가장 마지막 offset부터 - ..

카프카 consumer로 입수를 진행하다보면 예상치 못하게 데이터가 중복입수 또는 유실될 가능성이 있습니다. - Kafka broker 이슈 - Network 이슈 - Consumer application 이슈 위와 같은 이슈가 발생했을 경우 이슈가 발생했던 시점보다 더 이전의 데이터부터 입수를 진행해야하는데 이때 offset을 지정해야합니다. offset을 지정하는 방법은 아래와 같습니다. 1) Consumer 생성 Consumer 생성은 재입수 하고자 하는 Consumer에 지정하여 신규로 생성. 2) Offset 지정(by console shell script) ./kafka-consumer-groups shell을 통해서 offset을 reset할 수 있습니다. offset reset 옵션: --..

AvroFlumeEvent포멧은 아래와 같은 특징을 가진다. - Header : Map - Body : ByteBuffer AvroFlumeEvent포멧을 사용하기 위해서 필요한 dependency dependencies { compile 'org.apache.flume:flume-ng-core:1.9.0' } DeserializeValue method : private static Event deserializeValue(byte[] value) throws IOException { Event e; DatumReader reader = new SpecificDatumReader(AvroFlumeEvent.class); ByteArrayInputStream in = new ByteArrayInputStr..

Docker로 감싼 application이 host machine에 떠 있는 web application에 접근하려고 하면 어떤 현상이 일어날까요? 호출하기 위해 시도한 방법은 아래와 같습니다. 시도방법 : Docker application(alpine linux)에서 http://localhost:3000 호출 결과 : connection refuse 발생 사유 : Docker application이 자기자신의 docker container내부에 떠 있는 localhost를 호출함(not host machine) 당연히 Docker container내부의 localhost를 호출하다보니 timeout으로 인해 connection refuse가 일어날 수 밖에 없습니다. 이를 해결하기 위한 방법은 Doc..

맥북에서 테스트를 위해 local hadoop을 띄우기 위해 테스트를 하다보면 아래와 같은 오류를 발생할 때가 있습니다. $ /usr/local/Cellar/hadoop/3.1.2/sbin/stop-all.sh WARNING: Stopping all Apache Hadoop daemons as a1003855 in 10 seconds. WARNING: Use CTRL-C to abort. Stopping namenodes on [localhost] localhost: ssh: connect to host localhost port 22: Connection refused Stopping datanodes localhost: ssh: connect to host localhost port 22: Conne..

Kafka burrow를 통해 모니터링 하다보면 더이상 모니터링 해도 되지 않는 consumer group이 남아있는 경우가 있습니다. 이런 경우에는 burrow의 http endpoint를 통해서 특정 consumer group을 제거하여 모니터링 대상에서 제거할 수 있습니다. 이번 포스팅에서는 어떻게 삭제하는지 알려드리도록 하겠습니다. Burrow에서 Consumer group 제거 URL path DELETE /v3/kafka/(cluster)/consumer/(group) CURL 예제 만약 cluster이름이 dev이고 consumer group이 di-test 라면 아래와 같이 작성합니다. curl -XDELETE http://localhost:8000/v3/kafka/dev/consumer/..

Amazon MSK(Managed Streaming for Apache Kafka)는 AWS에서 제공하는 완전 관리형 apache kafka 서비스입니다. 기존에 on-promise에서 사용하던 혹은 EC2로 관리하던 Apache kafka를 SaaS형태로 사용할 수 있습니다. Apache kafka의 특정 버젼을 그대로 사용할 수 있기 때문에 vanila apache kafka의 버젼별 api spec을 따라서 사용할 수 있습니다. 그럼 이번 포스트에서는 AWS MSK cluster를 직접 구성해보고, producer/consumer을 만들어 테스트해보겠습니다. MSK Cluster 생성 Cluster을 생성하기 위해서 AWS CLI를 사용하거나 혹은 AWS console을 사용할 수도 있습니다.이번 ..

티스토리블로그는 js, css, html 등 다양한 확장기능을 가지고 있습니다. 이러한 기능을 토대로 블로그를 더욱 멋지게 꾸밀 수 있는데요. 이번 포스팅에서는 블로그에 눈이 내리게 하는 효과를 적용하는 예제를 설명드리도록 하겠습니다. 준비물 준비물은 아래와 같습니다. 약간의 웹개발 지식과 시간을 약간 가미하여 눈내리는 블로그를 만들어 봅시다. - 티스토리 블로그 - 간단한 js, html 지식 - 5분 적용방법 1) snow.js 다운로드 적용할 javascript는 zmfe유저의 snow.js 입니다. fure javascript와 canvas로 구현한 오픈소스입니다. - url : https://github.com/zmfe/snow.js 위 url에서 source를 다운로드 할 수 있습니다. 그러나..

실시간으로 비디오 및 데이터 스트림을 손쉽게 수집, 처리 및 분석 솔루션 Amazon Kinesis를 사용하면 실시간 스트리밍 데이터를 손쉽게 수집, 처리 및 분석할 수 있으므로 적시에 통찰력을 확보하고 새로운 정보에 신속하게 대응할 수 있습니다. Amazon Kinesis 기능 - Kinesis Video streams : 비디오 스트림을 캡처 처리 및 저장 - Kinesis Data streams : 데이터 스트림을 캡쳐, 처리 및 저장 - Kinesis Data firehose : 데이터스트림을 AWS데이터 스토어로 로드 - Kinesis Data analytics : SQL 또는 Java를 통해 스트림 데이터를 분석 이번 포스팅에서 주요하게 볼 서비스는 Data streams 입니다. Data s..

실시간으로 비디오 및 데이터 스트림을 손쉽게 수집, 처리 및 분석 솔루션 Amazon Kinesis를 사용하면 실시간 스트리밍 데이터를 손쉽게 수집, 처리 및 분석할 수 있으므로 적시에 통찰력을 확보하고 새로운 정보에 신속하게 대응할 수 있습니다. Amazon Kinesis 기능 - Kinesis Video streams : 비디오 스트림을 캡처 처리 및 저장 - Kinesis Data streams : 데이터 스트림을 캡쳐, 처리 및 저장 - Kinesis Data firehose : 데이터스트림을 AWS데이터 스토어로 로드 - Kinesis Data analytics : SQL 또는 Java를 통해 스트림 데이터를 분석 이번 포스팅에서 주요하게 볼 서비스는 Data firehose 입니다. Data ..

빅데이터를 다루다보면 다양한 file format을 만나게 됩니다. 오늘 포스팅에서는 file format들에 대해 알아보도록 하겠습니다. Delimiter separated file 가장 많이 쓰이는 raw text기반의 구분문자로 이루어진 배열입니다. 아래와 같이 평문으로도 저장, 조회할 수 있습니다. 아래는 콤마(,)로 구분자를 지정한 CSV(comma-separated values)의 예제를 보여줍니다. 직업,이름,날짜,성별,지역 학생,David,20190204,M,USA 사업가,James,20180305,M,Canada 비슷한 포멧은 아래와 같은 방식들이 있습니다. - TSV : tab separated value - SSV : space separated value 위와 같은 방식들을 모두 합..

MirrorMaker 2.0은 KIP-382로 등록되어 있던 주제였으며, 2018년 10월 12일에 최초로 jira(KAFKA-7500)가 생성되었습니다. release될 때까지 약 1년이 걸렸는데요. 이번 포스팅에서는 MirrorMaker2.0이 왜 나왔고, 어떤 기능을 포함하고 있는지 살펴보겠습니다. 개발동기 MirrorMaker1은 꽤 오랜기간 사용되어 왔지만 몇가지 이슈가 있었습니다. Legacy MirrorMaker1의 문제점: - Topic들이 kafka default configuration기준으로 생성되었기 때문에 repartition하기 위해 재작업이 필요함 - ACL, configuration의 변경이 source/sink cluster사이에 sync되지 않음 - Record들은 Def..

그라파나는 모니터링 대시보드로 많이 쓰이는 웹 애플리케이션입니다. 다양한 기능을 제공하는데 그 중 하나는 Alert기능이 있습니다. 이 기능에는 graph이미지를 추가로 전송할 수 있는데, slack, email 등에 noti를 보낼때 매우 유용하게 사용할 수 있습니다. 이미지를 전달하기 위한 Redering은 grafana의 Grafana Image Renderer플러그인이 chromium(headless browser)을 통해서 수행합니다. Grafana Image Renderer을 사용하기 위해서는 몇가지 제한사항과 이슈에 대해 확인해야하는데 이 부분에 대해 정리하고자 합니다. Grafana Image Renderer 플러그인 사용 방법 1) grafana-cli로 설치 $ grafana-cli p..

셔터스톡에서 2020 컬러트랜드를 발표하였습니다. https://www.shutterstock.com/blog/trends/ko/2020-color-trends 스펙트럼은 아래와 같습니다. 러시라바 - #FF4500 아쿠아민트 - #7FFFD4 팬텀블루 - #191970 이외에도 여러 지역별 색상을 소개하였습니다. 각 컬러의 color조합을 보고싶다면 아래 링크에서 확인할 수 있습니다. https://www.shutterstock.com/colors 100s of color combinations, schemes, palettes - Shutterstock Bring your project to life with a full spectrum of beautiful colors for any color p..

KafkaSource는 스파크의 Structured Streaming에서 Apache kafka를 data source로 사용하기 위한 목적이다. 이 library의 source는 아래에서 확인할 수 있다. Spark Kafkasource : https://github.com/apache/spark/blob/master/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala 2개의 핵심 function을 아래와 같이 정리할 수 있다. - getOffset() : KafkaOffsetrReader를 사용하여 가장 최근의 offset을 가져온다. - getBatch() : offset의 처음부터 끝까지에 존재..

Faust는 Windowing을 쉽게 지원할 수 있는 기능을 가지고 있다. Windowing을 통해 지난 10분간 데이터의 분석 혹은 매 5분마다 1시간 간격의 데이터 분석과 같은 내용을 수행할 수 있다. 이번 포스팅에서는 Windowing을 사용하는 방법에 대해 알아보고자 한다. Faust는 Hopping, Tumbling window를 지원한다. Tumbling Windows Tumbling window는 중복되지 않은 데이터에 대해 일정간격으로 분석할때 사용된다. from dataclasses import asdict, dataclass from datetime import timedelta import json import random import faust @dataclass class Clic..

DateTimeFormatter는 Java에서 datetime을 표시할때 가장 많이 사용하는 코드중 하나입니다. 보통 아래와 같은 코드로 표기하여 사용하는데요. LocalDate m = LocalDate.of(2018,12,31); String str = m.format(DateTimeFormatter.ofPattern("yyyy-MM"); //2018-12 선언한 pattern에 따라 LocalDate가 변경되어 String으로 반환되는 모습을 볼 수 있습니다. 위 pattern은 아래와 같은 특징을 가지는데요. 위 특징에 따라 보통 년도를 표시할때 yyyy를 사용하곤 합니다. 그런데 stackoverflow의 이 질문(링크)에서는 yyyy대신 uuuu를 사용하라고 합니다. 왜일까요? 서기, 기원전 표..

Faust에는 Table이라는 개념이 있다. Table을 생성하면 스트림데이터와 함께 사용 될 수 있다. from dataclasses import asdict, dataclass import json import random import faust @dataclass class ClickEvent(faust.Record): email: str timestamp: str uri: str number: int app = faust.App("exercise6", broker="kafka://localhost:9092") clickevents_topic = app.topic("com.udacity.streams.clickevents", value_type=ClickEvent) # # TODO: Define a..

Faust에는 중요한개념 2가지 Processor과 Operation이 있습니다. Processors 스트림데이터는 끝없는 데이터의 연속적인 흐름입니다. 1개 이상의 Processor들은 callback형태로 동작하게 됩니다. Faust 기반의 application에서 동작하며, function을 사용하여 추가 library등을 조합하여 사용할 수도 있습니다. def add_default_language(value: MyModel) -> MyModel: if not value.language: value.language = 'US' return value async def add_client_info(value: MyModel) -> MyModel: value.client = await get_http_..

Faust는 python기반 스트림프로세싱을 위한 library입니다. 아래 코드는 kafka로부터 데이터를 가지고 오는 code입니다. Producer from dataclasses import asdict, dataclass import json import faust # 모듈 import @dataclass # 데이터 직렬화를 위한 json dataclass 선언 for faust class ClickEventSanitized(faust.Record): timestamp: str uri: str number: int app = faust.App("exercise3", broker="kafka://localhost:9092") # 카프카 브로커 clickevents_topic = app.topic("..

# Python Streams ٩(◕‿◕)۶ # Forever scalable event processing & in-memory durable K/V store; # w/ asyncio & static typing. import faust Faust는 python의 stream processing library이다. kafka stream의 python버젼이라고 볼수 있습니다. Faust는 stream processing과 event processing 둘다 지원합니다. DSL언어를 사용하지 않고 pytthon library로서 동작합니다. 그렇기 때문에 Flask, NumPy, PyTorch 등과 함께 사용 할 수 있습니다. - Faust url : https://faust.readthedocs.io..

데이터 스트리밍 이론 목차 - Strategies for Application Design - Combining Streams - Filtering Streams - Remapping Streams - Aggregating Streams - Handling Time and Windowing - Streams vs Tables 들어가기 전에 -Join (Streams) 1개 이상의 output stream의 조합. 보통 연관된 key로 조합하는 것 -Filtering (Streams) 데이터 스트림에서 일정 조건에 따라 데이터를 제거하는것 -Aggregating (Streams) 데이터 스트림에서 더하거나 빼거나 key를 기반으로 group을 만드는 것 -Remapping (Streams) 데이터 스트림에..

Udacity의 Data streaming Nanodegree program을 듣고 난 후기를 공유하고자 합니다. Nanodegree program은 udacity의 certification program입니다.다. 강의를 모두 수강하고 난 뒤에는 Nanodegree(나노디그리) 라고 하는 학위는 아니고 증명서를 발급받을 수 있습니다. 이번에 들은 강의는 Udacity의 데이터스트리밍 강의입니다. 기본적으로 2달의 수강기간을 예상치로 잡고 있으며 Python, SQL, ETL에 대한 기본적인 개념을 가지고 있어야만 수강하는데 문제가 없다고합니다. 가격은 1달에 한화 약 47만원입니다.(2달에 약 80만원) 데이터 스트리밍은 현대의 빅데이터 비즈니스모델을 실시간으로 분석하고 처리하기 위해 사용됩니다. 이 ..

TED는 세상을 좀 더 나은 곳으로 만들기 위해 고민하는 사람들의 학술장의 중 하나입니다. TED는 여러 좋은 강의들이 많습니다. 대부분 영상이 영어자막을 지원하며 유튜브, 네이버TV 기타 여러 스트리밍에서도 무료로 볼 수 있습니다. 대부분 매우 좋은 퀄리티의 강의로서 짧으면서도 명료하기 때문에 영어공부를 겸하기에 아주 좋습니다. TED를 통해 영어자막으로 공부하려고 하다보니 반복재생이나 이미 지나간 자막에 대해서 한번더 볼려고 하면 다시 영상을 되돌려야 하기도 합니다. 웨일 브라우저에서 TED를 공부하기에 좋은 앱이 있어서 소개하고자 합니다. 안봐도 비디오 안봐도 비디오는 웨일브라우저의 확장앱입니다. 동영상기반 일부 웹페이지와 연동되어 사용가능하며 반복재생 등과 같은 기능을 통해 학습이 가능합니다. 다..

아래 포스트는 confluent 블로그 글을 토대로 제 의견과 함께 정리한 글입니다. 위 블로그글은 Gwen Shapira(하둡 애플리케이션 아키텍처, 카프카핵심가이드 저자이자 confluent PM)이 작성한 글입니다. 쿠버네티스위에 카프카를 올려야하나? 쿠버네티스를 사용하는 주요 이유 2가지는 아래와 같습니다. - 개발자와 운영자 모두에게 workflow의 효율을 높임으로서 생산성이 높아짐 - "bin packing(1개 가상/물리 장비에서 여러 application을 돌리는 것)"을 통해 리소스 관리 효율화 가능 그런데, 카프카는 운영하기 그렇게 어렵지 않으며, 가끔 node(가상/물리 장비)의 모든 resource를 필요할때가 있습니다. 그렇기 때문에 쿠버네티스 위에서 카프카를 운영하는 것은 그닥..