Logstash에 API 이용해서 데이터 가져오기

이전에 파일을 이용해서 지하철 위치 정보를 가져오도록 하였는데, 이번에는 실시간으로 데이터가 변경되는 지하철 도착 정보 데이터를 API를 통해 가져오도록 할 것이다.

로그스태시의 기본 템플릿 구조는 아래와 같다.

input{
	{ 입력 플러그인 }
}

filter{
	{ 필터 플러그인 }
}

output{
	{ 출력 플러그인 }
}

참고로 logstash의 플러그인은 공식 사이트에 필요한 것을 찾아보도록 하면 된다..
https://www.elastic.co/guide/en/logstash/7.16/index.html

Conf 파일 작성

Input

가져오고자 하는 데이터는 https://data.seoul.go.kr/dataList/OA-12764/F/1/datasetView.do# 였다. 샘플 URL을 보면

다음과 같이 URL의 마지막에 조회하고자 하는 지하철 역명이 들어간다. 회사 지하철이 궁금하기에 회사 근처 역으로 설정하였는데, 한글 그대로 로그스태시로 가져간다면 문제가 생겨서 인코딩해주었다.
변환은 변환 사이트 에서 URL을 UTF-8로 Encode 해주었다.

input {
  http_poller {
    urls => {
      url => "http://swopenapi.seoul.go.kr/api/subway/424776766b6a796d3533646c636850/json/realtimeStationArrival/0/5/남성"
    }
    request_timeout => 60
    schedule => { cron => " * 0/5 * * * *"} # 5분마다 실행 
    codec => "json"
  }
}

• http_poller: 호출하고자 하는 API URL을 작성해주고, schedule에서 cron 표현식을 이용해 주기적으로 수행하도록 한다.
• cron: cron 표현식에 대해 잘 정리된 블로그가 있어 참고하였다. cron 표현식

filter

API로 데이터를 가져오게 된다면 위와 같이 realtimeArrivalList 리스트 형식의 데이터라 문제가 생겼었다. 따라서 split 필터를 이용해 배열 필드를 개별 이벤트로 분할하도록 한다.
또한 필요없는 데이터들을 remove_field를 이용해 지워주었다.
Split Filter 자세한 내용

filter{
  split {
    field => "realtimeArrivalList"
  }
  
  mutate {
    remove_field => ["[realtimeArrivalList][subwayNm]", "[realtimeArrivalList][statnFid]", 
    "[realtimeArrivalList][arvlMsg3]","[realtimeArrivalList][arvlMsg2]","[realtimeArrivalList][btrainSttus]"
    , "[realtimeArrivalList][arvlCd]" , "[realtimeArrivalList][beginRow]","[realtimeArrivalList][statnTid]", "[realtimeArrivalList][bstatnNm]"
    ,  "[realtimeArrivalList][statnList]", "[realtimeArrivalList][selectedCount]", "[realtimeArrivalList][endRow]", "[realtimeArrivalList][totalCount]"
    , "[realtimeArrivalList][pageRow]", "[realtimeArrivalList][curPage]", "[realtimeArrivalList][trainCo]", "[realtimeArrivalList][ordkey]"
    , "[realtimeArrivalList][statnId]", "[realtimeArrivalList][barvlDt]", "[realtimeArrivalList][subwayList]", "[realtimeArrivalList][bstatnId]"]
    remove_field => ["errorMessage","@timestamp","@version"]
  }
}

output

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "realtime-subway"
  }
  stdout{
    codec => rubydebug
  }
}

output은 파일에서 한 것처럼 index만 바꿔서 가져오도록 하였다.

Kibana devtools로 확인

GET realtime-subway-test/_search

잘 들어간 것을 확인할 수 있다 !!