잼의 이것저것

input { kafka { id => "input_kafka" bootstrap_servers => "source1:9092,source2:9092" group_id => "logstash-consumer" topics_pattern => "^test_topic" partition_assignment_strategy => "cooperative_sticky" auto_offset_reset => "latest" codec => json decorate_events => true } } filter{ mutate{ remove_field => ["@version","@timestamp"] //@version, timestamp 제거 } } output { kafka { bootstrap_servers =..

목차 [개요] MSSQL의 데이터를 Redis로 Sync하고 싶은데, 이 때 logstash를 활용해보려고 한다. update일자를 기준으로 최신데이터가 INSERT되거나 UPDATE되면 이를 logstash가 감지해서 Redis에 계속해서 업데이트해주는 파이프라인을 구축해보는 것이 목표. [로컬 세팅: Docker container로 MSSQL, Redis 구동] Kafka도 활용할 수 있을 것 같아 Kafka도 추가해놓았다. https://github.com/sjm767/docker-compose-storage/blob/main/docker/mssql-redis-sync/docker-compose.yml GitHub - sjm767/docker-compose-storage: 상황별로 필요한 docke..

[준비] 윈도우에서 WSL2 설치하기: https://gaesae.com/161#google_vignette 윈도우10 WSL2 설치하기, WSL2 우분투 설치하기 WSL이란? Linux용 Windows 하위 시스템을 사용하면 개발자가 기존 가상 머신의 오버헤드 또는 듀얼 부팅 설정 없이 대부분의 명령줄 도구, 유틸리티 및 애플리케이션을 비롯한 GNU/Linux 환경을 수정하 gaesae.com 우분투(WSL2사용 )환경 기준으로 진행하였다. kubectl 설치: https://kubernetes.io/docs/tasks/tools/install-kubectl-linux/ Install and Set Up kubectl on Linux Before you begin You must use a kubect..

MySQL에 적재되는 데이터를 Elasticsearch에 복제하여 사용하려는 요구가 있다. 다양한 방법이 있는 듯하다. logstash input - jdbc plugin output - elasticsearch plugin fluentd input - mysql-query, mysql-prepared-statement etc.. output - elasticsearch 이 중 fluentd에 mysql-replicator가 있어서 이 부분에 대해서 구동해본 내용 정리. [mysql-replicator?] github: https://github.com/y-ken/fluent-plugin-mysql-replicator GitHub - y-ken/fluent-plugin-mysql-replicator: F..

[CDC?] 데이터베이스 내 데이터 변경을 식별하여 필요한 후속처리 (데이터 전송/공유)를 자동화하는 기술 or 설계기법 실시간 데이터를 기반으로하고 싶을 때 폭넓게 활용된다. 흔히 데이터 백업이나 통합 작업을 할 경우 방대한 데이터를 다뤄야하는데, 원본 소스 데이터 중에 최근 변경된 데이터들만 골라내게 되면 다른 시스템에서 조회할 때 시스템로드도 줄이고 전체적인 생산성 향상도 노릴 수 있다. [구현 기법] Timetamp on Rows 변경인지를 위한 timestamp 컬럼을 두고 이 값보다 더 최근값이 있으면 레코드가 변경된 것으로 식별한다. Version Numbers on Rows 변경인지를 위한 version 컬럼을 두고 이 값보다 최근값이 있으면 레코드가 변경된 것으로 식별한다. Status ..

로그를 수집하고 분석할 때 ELK Stack은 매우 유용하게 사용되어 왔다. fluentd는 Docker나 kubernetes 환경과 같은 마이크로 서비스의 로깅에 인기를 끌고 있다. (fluentd가 CNCF의 일부이고, Treasure Data (고객데이터플랫폼 구축 클라우드 서비스?) 를 기반으로 구축되었기 때문) 어떤 사례에 어떤 도구를 사용하는 것이 좋을지 비교해본다. [Logstash vs Fluentd 비교] 항목 Logstash Fluentd 비고 코드 언어 JRuby (Java 필요) CRuby (Java 필요 X) 오픈 소스 Apache 2.0 라이센스 오픈소스 Apache 2.0 라이센스 오픈 소스. CNCF. 이벤트 라우팅 if-else 조건 기반 태그 기반 이벤트에 태그를 지정하는..

fluentd를 통한 데이터 파이프라인 구축을 위해서 로컬환경에서 fluentd - kafka로 이어지는 파이프라인을 구축해본다. [Docker Container로 간단히 구동하기] Docker-Image Pull docker pull fluent/fluentd:edge-debian Fluentd 설정 파일 생성 http input을 stdout으로 출력하는 형식으로 테스트한다. $ sudo vim fluentd.conf @type http port 9880 bind 0.0.0.0 @type stdout Docker Run # RUN $ sudo docker run -d --name fluentd -p 9880:9880 -v $(pwd):/fluentd/etc fluent/fluentd:edge-debi..

[Fluentd?] 로그 수집기 (보통 로그 수집의 목적으로 사용하나 다양한 데이터 소스로부터 데이터를 받아 처리할 수 있음) C와 Ruby로 개발됨. fluentd로 전달된 데이터는 tag, time, record (JSON)으로 구성된 이벤트로 처리됨. 원하는 형태로 가공 후 다양한 목적지 (ES, Kafka etc..)로 전달 될 수 있음. Filter, Buffer, Parser를 직접 설정할 수 있는 것이 가장 큰 장점 데이터 유실을 막기 위해 메모리와 파일 기반의 버퍼 시스템을 사용. Failover를 위한 HA구성도 가능. Use-Cases Application Log 수집 mongodb와 같은 DB에 데이터 수집 스트림 데이터 처리 Tag fluentd에서는 Tag를 통해 이벤트가 흘러가면..

[Why Streaming?] 데이터를 처리하는 역사를 따라가보면 Data Warehouse - Batching - Streaming 순으로 발전해왔다. Data Warehouse 클라우드 이전 시절에는 Data Warehouse만을 사용하였고, 이 공간 자체는 한정적이지 엄격한 schema를 기반으로 분석을 위한 데이터만 추출하여 저장하는 방식으로 사용하였다. 그리고 데이터를 실제로 사용하는 시점은 데이터가 모두 이관된 이후 BI툴을 이용한 접근만이 가능하였다. Batch 하둡의 Batch 시스템이 나오고나서 다양한 schema를 사용할 수 있었지만 여전히 배치가 끝난 후 데이터에 접근할 수 있었다. Streaming 다양한 schema 뿐만 아니라 Ingestion (데이터를 빼오는 시점)에 바로 데..

개인적으로 생각하는 Source, Sink별 사용할 만한 Connector를 정리해봄. Source Sink 용도 Tool 비고 참조 Kafka Kafka 기존 토픽 메시지 가공 후 새로운 토픽 생성 Ksql, Flink 스트림 처리 Kafka Elasticsearch 데이터 이관 logstash kafka input: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html output: https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html Kafka Mongodb 데이터 이관 logstash mongodb input은 없음. ka..

[Apache Flink?] 분산 스트림 처리를 하는 엔진으로 "unbounded", "bounded" 데이터에 대해 상태가 있는 (stateful) 스트림 처리를 하는데 적합한 프레임 워크 비슷한 프레임워크들 중에 Spark, Kafka Streams, Storm 등이 있음 Spark: 마이크로 배치 Kafka Streams: Data Source로 Kafka만 지원함. Storm: Exactly Once 처리를 지원하지 않음. [Unbounded(무한한)와 Bounded(경계있는) 데이터의 처리 ] 모든 종류의 데이터는 이벤트 스트림으로 생성될 수 있다. ex) 카드거래, 센서측정, 로그, 웹 사이트의 사용자 행동 이러한 데이터는 Unbounded 혹은 Bounded 스트림으로 처리될 수 있다. Un..

목차 [개요] 로컬환경에서 Logstash를 활용하여 Mongodb 데이터를 Kafka로 이관하는 작업을 검증해본다. Source: Mongodb Sink: Kafka [준비물] Docker (Docker-compose 필요) Mongodb, Kafka 등을 컨테이너로 구동할 Docker-Compose.yml 파일 Datagrip: Mongodb에 데이터를 넣고 확인할 도구 Kafka-ui: 카프카 어드민 Logstash [작업순서] Docker-Compose.yml 파일 작성 mongodb, zookeeper, kafka, kafka-ui 구동 Datagrip으로 mongodb 접속 database 생성 collecction (table) 생성 document (row) 생성 logstash 세팅 mo..

목차 [Logstash란?] 데이터 파이프라인 구조에서 데이터의 집계, 변환, 저장 역할을 수행함. Logstash는 형식이나 복잡성과 관계 없이 데이터를 동적으로 수집, 전환, 전송할 수 있다고 한다. grok을 이용해 비구조적 데이터에서 구조를 도출하여 IP 주소에서 위치 정보 좌표를 해독하고, 민감한 필드를 익명화하거나 제외시키며, 전반적인 처리를 손쉽게 해주는 역할을 한다. Components Source : 모든 형태, 크기, 소스의 데이터 수집 데이터의 매우 다양한 시스템에 매우 다양한 형태로 보관되어 있다. Logstash는 이같이 다양한 시스템에서 데이터를 가져오는 다양한 소스를 지원한다. Filter : 데이터 이동 과정에서의 구문 분석 및 변환 Logstash의 필터는 데이터가 소스에서..

[데이터 인프라의 목적] 비즈니스 리더들의 결정을 도와주기 서비스나 제품을 데이터의 도움을 받아 발전시키기 데이터 웨어하우스? 데이터 인프라의 기초는 각종 운영시스템이다. 이 데이터들을 분석하고 싶으면 각각의 분석 도구를 붙여아 한다. 어느 순간부터는 통합된 데이터 분석 환경이 필요하게 되었다. 그래서 다양한 소스로부터 데이터를 저장하는 DataWarehouse라는 곳이 필요하게 됨. 데이터 유형의 차이 운영시스템(Normalized Schema): 모든 데이터를 쪼개서 관리함. Warehouse(Dimensional Schema): 분석을 위한 단순한 데이터들만 관리함. [ETL, ELT] ETL이란 데이터의 추출(Extract), 변환(Transform), 로드(Load)를 나타내며 조직에서 여러 시..