Logstash 실시간 데이터 처리

이전 테스트 데이터 도큐먼트 상태 이미지

 

우선 이전 포스팅인 "날씨" 키워드를 사용해서 데이터를 수집한 결과 위의 이미지와 같이 데이터가 저장되는 것을 알 수 있었다.  여기서 내가 필요한 부분을 한번 정리해서 나타내면 아래와 같다

*필요한 필드 또는 수정해야 하는 필드 정리

• message -> 기본적인 텍스트 데이터가 저장되는 부분이며 데이터를 보면 리트윗된 데이터는  RT @email 형식으로 내 프로젝트에 필요 없는 데이터 형식이 들어가 있으므로 이 부분을 정리해줄 필요가 있다. 추가적으로 뒤에 링크가 붙는 경우가 있는데 이 부분 또한 제거해줘서 순수한 텍스트 부분만을 정리해 data필드에 모은다.
• @timestamp -> 현재 timestamp시간은 UTC로 적용되어 있는데 한국시간은 UTC+9이므로 한국시간인 필드를 새로 추가해 줘야한다.
• 기본적으로 데이터가 중복으로 들어오는 경우가 있다. 이 부분을 fingerprint 필터를 사용해서 중복을 제거해줘야 한다
• 필요 없는 필드 -> [agent, event, log, ecs, host, version, tags] 현재 내가 하려는 프로젝트는 사실 텍스트와 발생한 시간만 있으면 할 수 있다고 생각되기 때문에 나머지 필드는 삭제한다.

 

*Logstash pipeline 코드

input { #beat에서 데이터 받아오기 
  beats {
    port => 5044
  }
}

filter {
  # 트위터 api 실행시 저장되는 로그 데이터 2줄 삭제
  if [message] =~ "Starting new HTTPS" or [message] =~ "GET /2/tweets/search/stream HTTP/1.1"{
    drop { } 
  } 
  #기본적으로 사용하지 않는 필터 삭제
  mutate {
    remove_field => ["tags", "log", "input", "agent","ecs", "@version", "event"]
  }
  # 리트윗 데이터 전처리 
  if [message] =~ "RT @"{ #리트윗 데이터인 경우 텍스트 부분만 data에 저장
    grok{
      match => {"message" => "(?<tmp1>RT+)\s@(?<tmp2>\w*):\s%{GREEDYDATA:data}"}
    }
    mutate{
    remove_field => ["tmp1", "tmp2"]
    }
  }
  else{ #리트윗 데이터가 아닌경우
    grok { #정상적인 텍스트면 바로 저장
      match => {"message" => "%{GREEDYDATA:data}"}
    }
    if [message] =~ "@"{ #혹시 태그+아이디가 있는경우에 아이디 제거
      grok{
        match => {"message" => "@(?<tmp>\w*)\s%{GREEDYDATA:data}"}
      }
      mutate{
        remove_field => ["tmp"]
      }
    }
  }
  #중복된 데이터 갱신하기 위한 hash값 생성
  fingerprint {
    method => "SHA1"
    source => "data"
    target => "fingerprint"
  }
  # 한국시간 필터 새로 추가
  ruby{ 
    code => "event.set('kr_time', LogStash::Timestamp.new(event.get('@timestamp')+(9*60*60)))" 
  }
}

output {
  #elasticsearh에 저장
  elasticsearch { 
    hosts => ["elasticsearch:9200"]
    index => "test"
    document_id => "%{fingerprint}" #document_id를 fingerprint 즉 텍스트의 hash값으로 설정하여 중복 방지
  }
}

필터링이 정상적으로 작동된 모습 <날씨 키워드로 테스트>

여기서 message 필드를 삭제하지 않고 보존한 이유는 혹시 필터링이 잘못되는 부분이 있으면 어떤 경우인지 파악하기 위해서 삭제하지 않았다. 여기까지가 데이터 정리한 부분이고 다음 포스팅에서는 ingest pipeline을 이용해서 텍스트가 긍정인지 부정인지를 판단하는 부분을 구현할 예정이다.

 

*지금까지의 소감

생각보다 구현하는데 많은 시간이 소요되고 또 많은 변수들이 있어서 구현하는데 조금 오래걸린감이 있다. 아직 많이 부족해서 그런것 같다. 하지만 이번에 처음으로 구현하면서 Logstash filter 부분을 많이 익히게 되어서 많은것을 얻은 기분인것 같다. 자격증 준비까지 하면서 하다보니 많은 시간을 투자하지 못하는게 너무 아쉽다