notfounduser's diary

Kafka 는 대용량 데이터를   위한   분산형 스트리밍 플랫폼입니다 . Kafka clustring Test 를   수행해   봅니다 . 전체Archtecture #1 ip setting 3대의 Centos 7.3 준비 server.1=192.168.0.145:2888:3888 server.2=192.168.0.174:2888:3888 server.3=192.168.0.175:2888:3888 #2 setting zookeeper & kafka  ## download zookeeper & kafka wget http://mirror.navercorp.com/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz tar zxvf kafka_2.12-2.2.0.tgz mv kafka_2.12-2.2.0 kafka configure zookeeper  (1,2,3 server)[root@kafka1 min]# vi kafka/config/zookeeper.properties # zookeeper servers server.1=192.168.0.145:2888:3888 server.2=192.168.0.174:2888:3888 server.3=192.168.0.175:2888:3888 ### configure kafka [root@kafka1 min]# vi kafka/config/server.properties broker.id=1 ## id를 server별로 설정 listeners=PLAINTEXT://:9092 zookeeper.connect=192.168.0.145:2181,192.168.0.174:2181,192.168.0.175:2181 delete.topic.enable=true #3 start zookeeper & kafka  #### start zookeeper serve

Kafka  + Python 연결 테스트 #1 Start zookeeper & kafka [root@zepp zookeeper-3.4.14]# vi conf/zoo.cfg [root@zepp zookeeper-3.4.14]# bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /home/min/zookeeper-3.4.14/bin/../conf/zoo.cfg Starting zookeeper … STARTED [root@zepp kafka_2.12-2.2.0]# bin/kafka-server-start.sh config/server.properties [2019-04-19 18:12:03,572] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$) [2019-04-19 18:12:04,104] INFO starting (kafka.server.KafkaServer) [2019-04-19 18:12:04,105] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer) [2019-04-19 18:12:04,128] INFO [ZooKeeperClient] Initializing a new session to localhost:2181. (kafka.zookeeper.ZooKeeperClient) #2 create topic [root@zepp kafka_2.12-2.2.0]# ./bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic suic

druid에 수집된 데이터셋을 superset에서 시각화를 통한 분석을 수행합니다. 전체 분석 flow #1 druid 수집 머신러닝을 활용한 빅데이터 분석 #3 에서 수집된 데이터셋의 확인 ## druid-kafka indexing list http://192.168.0.166:8081/#/indexing-service ## dsql에서 데이터셋 확인 select * from suicides4kafka 2 Druid-Superset Druid에서 수집된 데이터셋을 Superset과 연결하고 데이터 분석을 수행합니다. #2-1 Connect Superset-Druid Druid Cluster와의 연결을 위한 설정을 수행합니다. 연결이 완료되면 Datasources 스캔을 통해 Druid Datasources설정 메뉴에서 데이터셋 스키마를 확인 할 수 있습니다. #2-2 Analysis Superset에서 간단하게 수집된 데이터셋에 대한 내용을 확인해 봅니다. #3 Result 전체 flow Apachi nifi -> kafka -> druid -> superset

Apachi nifi에서 취득한 json 데이터를 kafka broker를 이용해 메세징 처리합니다. 이후, druid indexing처리를 이용하여 분석을 위한 데이터셋를 작성합니다. 전체 분석 flow #1 Kafka topic 만들기 이하 명령을 실행하여  suicide4  라는 카프카 항목을 만들고 여기에 데이터를 보내십시오 ## list kafka topic bin/kafka-topics.sh –list –zookeeper localhost ## delete kafka topic bin/kafka-topics.sh –delete –zookeeper localhost –topic suicides2 ## 위의 명령으로 삭제 되지 않을 경우, zookeeper shell을 이용하여 삭제 [zk: localhost:2181(CONNECTED) 2] ls /brokers/topics [suicides3, suicides4, __consumer_offsets] [zk: localhost:2181(CONNECTED) 3] rmr /brokers/topics/suicides3 [zk: localhost:2181(CONNECTED) 4] rmr /brokers/topics/suicides4 [zk: localhost:2181(CONNECTED) 5] ls /brokers/topics [__consumer_offsets] [zk: localhost:2181(CONNECTED) 6] ## create kafka topic bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic suicides4 #2 Druid Kafka ingestion Druid의 Kafka 인덱싱 서비스를 사용하여 #1에서 작성한 suicides4에서

Apachi nifi를 이용해서 csv파일을 로드하고 json형태의 데이타로 data transform을 구현합니다. 전체 분석 flow #1 Dataset 개요 https://www.kaggle.com/russellyates88/suicide-rates-overview-1985-to-2016 시간,장소, 연령 등의 요소가 포함된  데이터 셋을  분석하여  자살율 증가를 예방함을 목적으로 합니다.  country, year, sex, age group, count of suicides, population, suicide rate, country-year composite key, HDI for year, gdp_for_year, gdp_per_capita, generation (based on age grouping average). #2 Load Data with NIFI 프로세서 flow GetFile -> InferAvroSchema -> ConvertCSVToAvro -> ConvertAvroToJSON -> PublishKafka #2-1 GetFile Processor Apache NiFi 프로세서는 데이터 흐름을 만드는 블록입니다. 모든 프로세서는 출력 흐름 파일 생성에 기여하는 각각의 기능을 가지고 있습니다. 아래 이미지에 표시된 데이터 흐름은 GetFile 프로세서를 사용하여 한 sucides cvs 파일을 가져 와서 PutFile 프로세서를 사용하여 다른 디렉터리에 저장합니다. Input Directory, File Filter란에 수집대상 cvs파일의 Directory 및 파일명을 설정합니다. #2-2 InferAvroSchema Processor 입력 받은 콘텐츠에서 Avro 스키마를 자동으로 생성합