logstash에서 elasticsearch로 mysql 데이터 보내기

이번에는 logstash에서 elasticsearch로 mysql 데이터를 보내 보겠습니다!

1. use_bike.conf 파일 작성

schedulelinux

cron과 작성법이 동일합니다. 1시간에 한번 데이터를 받아오게 설정했습니다.
schedule => "0 * * * *"

filter

date {
    match => ["date", "yyyy-MM-dd HH"]
    timezone => "Asia/Seoul"
  }

한국시간이라고 말해주지 않으면 키바나에서 시각화를 할 때 +9시간이 더해져서 보이게 됩니다.

input {
  jdbc {
    jdbc_driver_library => "/usr/share/java/mysql-connector-java-8.0.29.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://rds주소:3306/db"
    jdbc_user => 'id'
    jdbc_password => 'pw'
    statement => "쿼리"
    schedule => "0 * * * *"
  }
}

filter {
  date {
    match => ["date", "yyyy-MM-dd HH"]
    timezone => "Asia/Seoul"
  }
}
output {
  elasticsearch {
    hosts => "13.125.224.72:9200"
    index => "use_bike"
 }
}

2. weather.conf 파일 작성

1번과 같습니다.
테이블 설정만 바꿔주면 됩니다.

input {
  jdbc {
    jdbc_driver_library => "/usr/share/java/mysql-connector-java-8.0.29.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://rds주소:3306/db"
    jdbc_user => id
    jdbc_password => pw
    statement => "쿼리"
    schedule => "0 * * * *"
  }
}

filter {
  date {
    match => ["date", "yyyy-MM-dd HH"]
    timezone => "Asia/Seoul"
  }
}
output {
  elasticsearch {
    hosts => "13.125.224.72:9200"
    index => "weather"
 }
}

3. use_bike index mapping

한글을 사용하려고 노리 토크나이저를 이용했습니다.

PUT use_bike
{
  "settings": {
    "analysis": {
      "analyzer":{
        "nori": {
          "tokenizer": "nori_tokenizer",
          "decompound_mode": "mixed"
        }
      }
    }
  }, 
  
  "mappings": {
    "properties": {
      "idx":{"type": "text"},
      "stationid":{"type": "text"},
      "stationname":{
        "type": "text",
        "fields": {
          "nori": {
            "type": "text",
            "analyzer": "nori"  
          }
        }
      },
      "racktotcnt": {"type": "integer"},
      "parkingbike": {"type": "integer"},
      "shared": {"type": "double"},
      "location": {"type": "geo_point"},
      "date": {
        "type": "date",
        "format": "yyyy-MM-dd HH"
      }
      
    }
  }
  
}

4. weather index mapping

PUT weather
{
  "mappings": {
    "properties": {
      "idx": {"type": "short"},
      "pty": {"type": "double"},
      "reh": {"type": "integer"},
      "rn1": {"type": "double"},
      "t1h": {"type": "double"},
      "uuu": {"type": "double"},
      "vec": {"type": "double"},
      "vvv": {"type": "double"},
      "wsd": {"type": "double"},
      "date": {
        "type": "date",
        "format": "yyyy-MM-dd HH"
      }
    }
  }
}

5. 파이프라인 설정

날씨와 자전거 두개의 파이프라인을 설정합니다.

- pipeline.id: bike
  path.config: "/usr/share/logstash/config/use_bike.conf"

- pipeline.id: weather
  path.config: "/usr/share/logstash/config/weather.conf"

6. 실행

./bin/logstash --path.settings /etc/logstash
이렇게 실행하면 로그스태시가 알아서 pipelines.yml파일을 읽게 되고. 안에 적힌 경로를 찾아갑니다.