[ELK] ELK활용 미니프로젝트 - ELK 하나씩 구성해가기

 

 

 

 

💻 ELK 하나씩 구성해가기

 

 

 

ELK 설치가 다 되었다는 전제 하에 ELK 하나씩 구성해보자~

 

설치에 관련해서는 이전에 발행했던 아래를 참고하면 된다!

2021.12.15 - [빅데이터/빅데이터 자료 공유] - [자료공유] Centos7 환경에서 Elastick stack 설치하기(elastic search, kibana, logstash, filebeats )

[자료공유] Centos7 환경에서 Elastick stack 설치하기(elastic search, kibana, logstash, filebeats )

 

 

나의 elk 논리 구성은 다음과 같다!!

 

 

데이터 분석 흐름을 풀어 설명해보자면 ,

Filebeat로 csv파일을 catch하여

전처리 CSV 파일을 Logstash로 data loading한다음

ElasticSearch로 data를 보내 분석한 후

Kibana를 통해 시각화 및 관리

 

터미널에서 실행하는 순서는 ( 순서 중요❗❗ )

elasticsearch 실행

kibana 실행

작성한 logstash conf파일 실행

filebeat 실행

 

---

 

🥕1.  나중에 filebeat가 catch하기 원하는 경로(path)에 전처리 파일 가져오기

- 원하는 경로:  /home/sonalim/elk 

🥕2. elasticsearch/bin 폴더에서 elasticsearch 실행

 

./elasticsearch

 

 

🥕3. kibana/bin 폴더에서 kibana 실행

./kibana

 

 

🥕4. logstash/config에 있는 logstash-sample.conf 을 복사하여서,

birth_year.conf(연도별)과 birth_gen.conf(세대별) 만들기

(logstash의 conf는 설정파일이다.) 

 

 

🥕5. 각 conf 파일 속 내용 저장하기!

- birth_year.conf(연도별)

input { beats { port => 5044 # beat에서 데이터를 받을 port지정 } } filter { mutate { # 실제 데이터는 "message" 필드로 오기 때문에 csv형태의 내용을 분할하여 새로>운 이름으로 필드를 추가해준다. split => [ "message", "," ] add_field => { "month" => "%{[message][0]}" "1985" => "%{[message][1]}" "1986" => "%{[message][2]}" "1987" => "%{[message][3]}" "1988" => "%{[message][4]}" "1989" => "%{[message][5]}" "1990" => "%{[message][6]}" "1991" => "%{[message][7]}" "1992" => "%{[message][8]}" "1993" => "%{[message][9]}" "1994" => "%{[message][10]}" "1995" => "%{[message][11]}" "1996" => "%{[message][12]}" "1997" => "%{[message][13]}" "1998" => "%{[message][14]}" "1999" => "%{[message][15]}" "2000" => "%{[message][16]}" "2001" => "%{[message][17]}" "2002" => "%{[message][18]}" "2003" => "%{[message][19]}" "2004" => "%{[message][20]}" "2005" => "%{[message][21]}" "2006" => "%{[message][22]}" "2007" => "%{[message][23]}" "2008" => "%{[message][24]}" "2009" => "%{[message][25]}" "2010" => "%{[message][26]}" "2011" => "%{[message][27]}" "2012" => "%{[message][28]}" "2013" => "%{[message][29]}" "2014" => "%{[message][30]}" "2015" => "%{[message][31]}" "2016" => "%{[message][32]}" "2017" => "%{[message][33]}" "2018" => "%{[message][34]}" "2019" => "%{[message][35]}" "2020" => "%{[message][36]}" } # 기본으로 전송되는 데이터 분석에 불필요한 필드는 제거한다. "message" 필드도 위에서 재가공 했으니 제거한다. remove_field => ["ecs", "host", "@version", "agent", "log", "tags", "input", "message"] }

# Kibana에서 데이터 분석시 필요하기 때문에 숫자 타입으로 변경해준다. mutate { convert => { "month" => "integer" "1985" => "integer" "1986" => "integer" "1987" => "integer" "1988" => "integer" "1989" => "integer" "1990" => "integer" "1991" => "integer" "1992" => "integer" "1993" => "integer" "1994" => "integer" "1995" => "integer" "1996" => "integer" "1997" => "integer" "1998" => "integer" "1999" => "integer" "2000" => "integer" "2001" => "integer" "2002" => "integer" "2003" => "integer" "2004" => "integer" "2005" => "integer" "2006" => "integer" "2007" => "integer" "2008" => "integer" "2009" => "integer" "2010" => "integer" "2011" => "integer" "2012" => "integer" "2013" => "integer" "2014" => "integer" "2015" => "integer" "2016" => "integer" "2017" => "integer" "2018" => "integer" "2019" => "integer" "2020" => "integer" } remove_field => [ "@timestamp" ] } } output { # 콘솔창에 어떤 데이터들로 필터링 되었는지 확인한다. stdout { codec => rubydebug } # 위에서 설치한 Elasticsearch 로 "birth_year" 라는 이름으로 인덱싱 한다. elasticsearch { hosts => ["http://localhost:9200"] index => "birth_year" } }

 

- birth_gen.conf(세대별)

input { beats { port => 5044 # beat에서 데이터를 받을 port지정 } } filter { mutate { # 실제 데이터는 "message" 필드로 오기 때문에 csv형태의 내용을 분할하여 새로>운 이름으로 필드를 추가해준다. split => [ "message", "," ] add_field => { "month" => "%{[message][0]}" "Y_Gen" => "%{[message][1]}" "Z_Gen" => "%{[message][2]}" "Gold_baby_Gen" => "%{[message][3]}" } # 기본으로 전송되는 데이터 분석에 불필요한 필드는 제거한다. "message" 필드도 위에서 재가공 했으니 제거한다. remove_field => ["ecs", "host", "@version", "agent", "log", "tags", "input", "message"] } # Kibana에서 데이터 분석시 필요하기 때문에 숫자 타입으로 변경해준다. mutate { convert => { "month" => "integer" "Y_Gen" => "integer" "Z_Gen" => "integer" "Gold_baby_Gen" => "integer" } remove_field => [ "@timestamp" ] } } output { # 콘솔창에 어떤 데이터들로 필터링 되었는지 확인한다. stdout { codec => rubydebug } # 위에서 설치한 Elasticsearch 로 "birth_year" 라는 이름으로 인덱싱 한다. elasticsearch { hosts => ["http://localhost:9200"] index => "birth_generation" } }

 

 

🥕6. logstash 폴더에서 작성한 logstash conf파일 실행

bin/logstash -f config/birth_gen.conf

 

 

🥕7. filebeat catch 경로(path)설정하기

filebeat.yml에서  paths는 해당경로의 모든 csv파일 읽을수 있도록 설정하기

 

 

🥕8. filebeat 폴더에서 filebeat.yml파일 실행

./filebeat -e -c filebeat.yml

 

 

🥕9. 인터넷을 켜고 5601포트로 Kibana 홈페이지 접속한다!

 

 

 

 

 

이 구성하는 과정이 생각보다 오래걸린다....

리눅스 명령어가 능숙하지 않으면 오래걸리는 작업!!

 

 

 

🤞🍀