기존에 작업을 python-kafka로 겨우 익숙해졌는데, 비동기적인 처리 방법이 필요하게 됐다. 그래서 찾아보니 비동기 코 루틴인 asyncio 방식이 필요하게 됐다. 처음에는 지나친 패키지였지만 asyncio 개념이 필요하게 돼서 다시 보게 되니 생각보다 간편했다. 그래서 찾은 것이 aiokafka이다! 일단 이걸 하게 된 이유는 다음과 같다. asyncio를 적극적으로 쓴다. python-kafka 보다 거의 동일한 형태라서 익숙하다. python-kafka 보다 좀 더 성능이 좋다고 한다. URL python-kafka는 속도를 중시하지 않았지만, 자바 클라이언트 변화에 빠르게 변화했다. Producer 뿐만 아니라 Consumer도 기능이 python-kafka 보다 훨씬 다양하다. batch ..

conda 나 virtualenv 같은 경우에는 환경을 여러 개를 만들 수 있다. 그렇다면 이러한 환경들을 여러 개로 다루면서, 사용하고 싶은 경우가 있다. 이 글에선 결론적으로 sh로 실행을 할 것이고, 그전에 일단 각각의 환경에 대해서 다른 버전을 제공하는지를 확인해보자. 일반적으로 환경에 접속을 하고 나서 실행을 한다. 보통은 activate를 하고 나서, 해당 파이썬에서 script를 실행한다. ## version check conda info --envs conda activate py36 python test.py 하지만 이런 경우에는 sh을 이용할 때 문제가 발생한다. 일단 sh에서 conda activate 를 하고 python을 실행해주면 될 것 같기도 한데...(테스트 안 해봄) 그래서..

폴더 복사를 하려고 하는데, 한번 하고 나니 폴더가 있다면서 복사가 되지 않았다. cp -r from_folder to_folder 이런식으로 진행을 하니 파일이 존재한다해서 안된다고 해서 방법을 찾아보니 다음과 같았다. cp -TRv from_folder to_folder -T : --no--target-directory -v : --verbose - 끝 -

jupyter notebook 키려다가 error_connection timed out 발생 port는 listen을 하지만, 켜지지는 않은 상태 확인해보니 그전에 firewall를 설치하면서 firewall가 작동해서 차단된 거였음 그래서 firewalld를 중지하니 작동 ## port 열려있는지 확인 netstat -tnlp ## 상태 확인 systemctl status firewalld ## 멈추기 systemctl stop firewalld ## port 확인 firewall-cmd --list-port ## 개별적으로 port 열기 firewall-cmd --permanent --zone=public --add-port=8888/tcp ## 재구동 firewall-cmd --reload ## 개..

## Producer 1 (topic odd) from confluent_kafka import Producer import numpy as np p = Producer({'bootstrap.servers': 'localhost'}) def delivery_report(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush(). """ if err is not None: print('Message delivery failed: {}'.format(err)) else: print('Message delivered to {} [{}]'.format(msg.topic(..

import os from kafka import KafkaConsumer class Commad : def __init__(self, path = "/usr/local/kafka/bin/" ) : self.consumer = KafkaConsumer( bootstrap_servers=['localhost:9092'], auto_offset_reset='latest' , # 'earliest', enable_auto_commit= True ,) self.path = path def show_topic(self,) : """ topic 보여주기 """ print("Topic : " , list(self.consumer.topics())) def create_topic(self, partition , rep..

가끔 코드를 숨길일이 있다. 그땐 이것을 사용하면 된다. 그러나 주의사항! 밑에 from~ 저 부분을 없애면 열수가 없다는 거 저 버튼이 사라지게 된다 HTML(''' ''') 그냥 없애고 싶으면 밑에 처럼 하면 될 듯? from IPython.core.display import HTML HTML(''' ''') 물론 div다 보니 다시 보이게 할 수 있긴 한 것 같다...ㄷㄷㄷ

## 키기 zookeeper -> server /usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties ## show topic list /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181 ## create the topic /usr/local/kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --rep..

## 모든 연결 출력 netstat -a ## tcp 또는 udp 연결만 표시 netstat -at netstat -au ## 목록을 빨리 출력하기위해 역 dns 질의를 끄기 netstat -atn ## listen 상태인 연결만 표시하기 netstat -tnl ## 프로세스 이름과 pid도 표시하기 apache나 nginx가 포트를 점유하고 있는지, 동작중인지 아닌지 확인하고 싶을 때 프로세스 이름과 pid 까지 보여준다면 더 유용할 것입니다. 프로세스 상세 정보 표시 옵션은 'p'입니다. 이 옵션은 root 권한을 가진 상태에서 실행해야 합니다. root 권한이 없을 경우 pid를 알아낼 수 없습니다. netstat -tnlp ## 커널 라우팅 정보 출력하기 netstat -nr ## 리눅스 로컬서버..

https://stackoverflow.com/questions/46491616/python-kafka-multiprocess-vs-thread Python Kafka multiprocess vs thread I can use KafkaConsumer to consume messages in separate threads. However, when I use multiprocessing.Process instead of threading.Thread, I get an error: OSError: [Errno 9] Bad file descriptor T... stackoverflow.com 같은 group_id를 가지면서( offset 순차적 ) 같은 토픽에서 정보 가져오기 이렇게 하면 얻는 효과는 더 빨..

Producer from time import sleep from json import dumps from kafka import KafkaProducer import numpy as np producer = KafkaProducer(bootstrap_servers=['localhost:9092'], key_serializer = None , value_serializer=lambda x: dumps(x).encode('utf-8')) ## for _ in range(100): value = np.random.normal(loc = 10 , scale = 20 ,size= 3).astype(str).tolist() values = ",".join( value ) data = {'number' : valu..

현재 카프카를 공부할 일이 있어서 원래는 그냥 인터넷에 있는 글들을 읽어서 파악하려고 했지만, 영어 바보인 관계로 두리뭉실하게 알고 있는 것 같아서 카프카, 데이터 플랫폼의 최강자 책을 읽어보려고한다. 책에서는 현재 자바/파이썬으로 코드가 조금 있는 것 같아서 더 적합한 것 같다. 다른 책들은 일단 도서관에 없었으므로... 일단 이 책도 좋은 것 같아서 시작한다! 더 읽다가 괜찮으면 사야겠다! ㅎㅎㅎ http://www.yes24.com/Product/Goods/59789254 카프카, 데이터 플랫폼의 최강자 데이터 플랫폼의 핵심 컴포넌트로 각광받고 있는, 이벤트 기반 비동기 아키텍처를 위한 고가용성 실시간 분산 스트리밍 솔루션 카프카(Kafka)의 모든 것!국내 최대 모바일 플랫폼 회사인 카카오에서 `..