지나가던 컴공생

- zookeeper 설정을 완료했으니 이제 kafka를 설치해보자. wget https://archive.apache.org/dist/kafka/3.3.0/kafka_2.12-3.3.0.tgz tar xvf kafka_2.12-3.3.0.tgz - kafka 실행 전에 broker.id 설정, zookeeper에 대한 설정과 listener설정을 해야한다. 이때 broker.id를 모두 다르게 설정해야한다. zookeeper.connect 설정시 마지막에 /test 와 같이 route를 넣으면, zookeeper의 root node가 아닌 child node에 카프카정보를 저장하게 되므로 유지보수에 이득이 있다고 한다. # /home/ec2-user/kafka_2.12-3.3.0/config/serve..

- 이제 도커가 아닌 AWS EC2 인스턴스를 발급받아서 Kafka를 운영할 것이다. - 본인은 EC2에서 t2.micro유형의 인스턴트를 3개 발급받았다. 발급받은 pem키를 원하는 디렉토리에 위치시킨 후 윈도우 유저이므로 OpenSSH를 사용하여 인스턴스에 접속한다(맥북 사고싶다 ㅠㅠ). - Open SSH의 설치는 링크를 참고하자. - 인스턴스에 "연결"을 클릭하면 SSH 클라이언트를 통해 연결하는 방법이 나와있다. 하지만 그냥 접속하면 아마 접근이 안 될 것이다. EC2 대시보드에 들어가 "보안" 에서 인바운드 규칙을 편집해야한다. 처음 생성했을때 default로 생성된 보안 그룹에 들어가 유형을 "SSH", 소스 내 IP로 설정하면 된다. 참고로 AWS 콘솔을 사용해 접근하려면 추가적으로 설정이..

- Smilegate korean_unsmile 데이터 링크 : https://github.com/smilegate-ai/korean_unsmile_dataset - 링크를 통해 가상환경에서 pip로 torch 설치 (놋북이라 GPU 없으므로 CUDA 없는 버전으로 설치) pip3 install torch torchvision torchaudio - transformers 및 기타 라이브러리 설치 pip install transformers pip install chardet - 아래 코드 테스트 from transformers import TextClassificationPipeline, BertForSequenceClassification, AutoTokenizer model_name = 'smile..

- 이제 클러스터에서 토픽을 생성해보자. KAFKA_ADVERTISED_LISTENERS에 post를 19091로 설정했기 때문에 bootstrap-server를 localhost:19091로 설정하고, "first-cluster-topic"이라는 이름으로 토픽을 생성했다. 파티션은 3개를 생성했고, replication-factor는 1로 설정했다. docker exec -it [컨테이너ID] kafka-topics --bootstrap-server=localhost:19091 --create --topic first-cluster-topic --partitions 3 --replication-factor 1 - kafdrop에서 확인해 보면, 아래와 같이 토픽이 잘 생성된 것을 볼 수 있다. - CL..

- 이제 도커를 사용해 Kafka 클러스터를 만들어 실제 분산환경과 같이 kafka를 사용해 볼 것이다. 우선 kafka와 zookeeper를 stop하고 netstat를 사용해서 제대로 작동 중지했는지 확인하자. - 이후 docker-compose.yaml 파일을 작성할 것이다. 아래와 같이 코드를 짜고 도커 데스크탑을 실행해준 다음, docker-compose up을 해주면 컨테이너가 올라온 것을 볼 수 있을 것이다. 이로써 zookeeper 클러스터를 생성하였다. version: '3' services: zookeeper: image: zookeeper:3.8 hostname: zookeeper ports: - "2181:2181" environment: ZOO_MY_ID: 1 ZOO_PORT: 2..

- 이제 Kafka를 사용하여 직접 프로그래밍을 해볼 것이다. 우선 링크에서 kafka를 다운로드하자. (kafka 3.4.0, Scala 2.13버전을 사용) (만약 jdk가 설치되어 있지 않으면 설치 후 환경변수 설정까지 해야함) - 압축을 풀고 "~/user" 디렉토리에 옮긴다. 이후 Kafka 설정 파일을 수정한다(./config/server.properties) 1. log.dirs를 원하는 디렉토리로 수정 2. host와 port를 설정 - 이제 Zookeeper를 실행하고 (port:2181) (kafka_test) C:\Users\subo0\kafka_2.13-3.4.0\bin\windows>zookeeper-server-start.bat ../../config/zookeeper.prope..

- 이전 포스트의 내용을 좀 더 자세하게 알아보자. - Kafka의 토픽은 파티션의 그룹이고, 디스크에는 파티션 단위로 저장된다. 각 파티션마다 메시지(commit log)가 쌓이게 되는데, 이 메시지는 정렬되어 있고 불변성을 가진다. 또한 파티션의 모든 기록들을 Offset이라는 ID를 부여받는다. - Kafka의 메시지는 Byte의 배열이며 String, JSON, Avro를 사용한다. Avro는 직렬화되어 성능을 높인 스키마를 표현하는 JSON이라고 생각하면 된다. 메시지의 크기에는 제한이 없지만 성능을 위해 작게(KB단위) 유지하는 것을 추천한다. - 데이터는 사용자가 지정한 시간(Retention Period)만큼 저장한다. 또한 토익별로 지정도 가능하다. Consumer가 데이터를 받아가고 나..

- 기존 스트리밍 프로세싱을 하려면 아래 그림과 같이 시스템을 더할수록 복잡해지는 문제가 있었다. 여기서는 여러가지 통신 프로토콜(HTTP, TCP, gRPC, MQ 등)을 지원해야 하고 각 데이터 포멧도 다르며, 무엇보다도 Point-of-failure가 많았다. 예를 들어 각 시스템 A,B,C의 신뢰도가 99%인 시스템을 묶으면 신뢰도가 97%까지 떨어진다. 이는 시스템이 커질 수록 신뢰도가 더 떨어지는 문제가 발생할 수 있다. - Kafka는 분산 스트리밍 플랫폼이다. Source 시스템(Producer)은 Kafka로 메시지를 보내고, Destination 시스템(Comsumer)은 Kafka로부터 메시지를 받는다. 또한 확장성이 있고, 장애 허용(fault tolerant)를 하며, 기존 시스템..

OLTP : - A계좌에서 B계좌로 1000원을 송금할 때 A에서 1000원을 빼고 B에 1000원을 추가하는 두 가지 작업을 모두 만족해야한다. 이 작업을 트랜잭션으로 묶어서 처리해주는 DB를 OLTP라고 한다. - 보통 수많은 트랜잭션이 서비스에서 일어나고, 빠른 프로세싱을 요구하기 때문에 정규화된 데이터로 처리한다. 또한 수많은 테이블로 이루어져있다. cf) OLAP는 데이터분석이나 데이터 웨어하우스에서 데이터를 처리하는 것과 비슷하다. 데이터의 볼륨이 크고, 복잡한 분석쿼리를 날리며 비정규화된 데이터로 이루어진다. CDC : - OLTP에서 일어나는 대부분의 행동은 데이터가 업데이트되는 일이다. OLTP의 DB는 트랜잭션을 기록하는데에만 집중되어 있고 그 트랜잭션 데이터를 다른 DB에 쓰지(W) ..

- https://school.programmers.co.kr/learn/courses/30/lessons/131532 - 레벨 : 4 더보기 distinct 유의,한 번 더 생각해볼 만한 문제 SELECT YEAR(a.sales_date) AS year, MONTH(a.sales_date) AS month, b.gender, COUNT(DISTINCT(a.user_id)) AS users FROM online_sale AS a INNER JOIN user_info AS b ON a.user_id = b.user_id WHERE b.gender IS NOT NULL GROUP BY year, month, gender ORDER BY 1, 2, 3 - https://school.programmers.co...