[Data 수집 및 연동] filebeat, kafka, logstash 사용해서 elasticsearch 데이터 업로드 하는 방법

💡flow 이해

 

1. 우선 내가 수집하려는 데이터의 구조는 master / detail 로 나누어져 있음

  - 인덱스 하나로 field라는 구분자를 사용하여 데이터 수집할 예정

2. filebeat.yml에서 data input, output 세팅해줌

3. kafka 토픽으로 위 파일을 수집하도록 함

4. logstash는 토픽과 연결하여 토픽 안에 있는 데이터를 수집하여 elastic에 뿌려줌

filebeat > kafka > logstash > elasticsearch

 

💡실습

 

1. filebeat.yml (permission X인 경우 sudo로 진행)

filebeat inputs 부분에 fields로 master / detail 구분해주도록 함

다른 output부분 주석처리해주고 kafka 부분만 주석제거하여 토픽 설정해주도록 함 

(hosts에 https붙이지 않도록 유의하기!)

저장 후 파일비트 실행하기 ./filebeat -e (sudo ./filebeat -e)

 

2. 카프카 폴더로 이동하여 서버 기동시키기

./bin/kafka-server-start.sh ./config/kraft/server.properties

 

3. topic list 확인하기

bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

apple-topic 들어온 걸 확인할 수 있음

그럼 저 토픽안에 filebeat로 쏜 데이터가 수집되어 있는 것!

 

아래 명령어 통해서 토픽 안 데이터들을 확인할 수 있음

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apple-topic

 

4. 로그스태시 폴더로 이동 (아래 명령어 입력)

vi kafka.conf

아래와 같이 입력해줌
input {
 kafka {
    bootstrap_servers => "localhost:9092"
    group_id => "logstash"
    topics => ["apple-topic"] 
    auto_offset_reset => "earliest"
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "apple_test"
    ssl_certificate_verification => false
    user => "elastic"
    password => "elastic"
  }
}

  

5. 로그스태시 실행 - elasticsearch로 인덱스 생성되어 추가됨

bin/logstash -f kafka.conf