이 글은 2015년 9월 10일 ObjectRocket.com/blog에 처음 게시된 내용을 기반으로 재구성했습니다.
새로 구축한 Elasticsearch® 인스턴스는 있는데, 검색하고 싶은 유용한 데이터가 CSV 파일에만 들어 있다면 어떻게 해야 할까요? 걱정할 필요 없습니다. Logstash®를 사용하면 거의 모든 형태의 데이터를 Elasticsearch 인덱스에서 쉽게 검색할 수 있는 형태로 변환할 수 있습니다.
시작하기 전에 준비할 것
이 예제를 따라 하려면 실제 데이터와 Unix® 계열 환경이 필요합니다. Windows®에서도 약간의 수정만 거치면 충분히 동작합니다. 여기서는 Davis Vantage Pro2® 기상 관측 장비에서 내보낸 .CSV 형식의 데이터를 활용해 새로운 인덱스를 만들어 보겠습니다.
먼저 로컬 파일에 저장된 수백만 건의 데이터가 다음과 같은 형태라고 가정해 보겠습니다.
$ head -3 /home/erik/weather.csv
HumOut,TempIn,DewPoint,HumIn,WindDir,RainMonth,WindSpeed,RainDay,BatteryVolts,WindChill,Pressure,time,TempOut,WindSpeed10Min,RainRate
76,78.0,78.227017302825,44,109,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:29Z,86.5,1,0.0
76,78.0,78.227017302825,44,107,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:45Z,86.5,1,0.0
76,78.0,78.32406784157725,44,107,2.0,0,0.0,1.236328125,90.83340000000001,29.543,2015-06-18T17:50:00Z,86.59999999999999,1,0.0
참고: 이 실습을 진행하려면 최소 하나 이상의 데이터 소스가 필요합니다.
Logstash 설치
데이터가 준비되었다면 본격적으로 시작할 수 있습니다. 먼저 Java가 설치되어 있는지 확인하세요.
$ java -version
openjdk version "1.8.0_51"
OpenJDK®, Oracle® 등 어떤 JVM(Java Virtual Machine)이든 무방합니다.
$ curl -O https://download.elastic.co/logstash/logstash/logstash-1.5.4.tar.gz
$ tar xfz logstash-1.5.4.tar.gz
$ cd logstash-1.5.4
$ mkdir conf
이제 설정 파일을 작성할 차례입니다.
입력(input) 섹션 정의
가장 먼저 input 섹션에서 Logstash에게 데이터의 위치를 알려줍니다.
input {
file {
path => "/home/erik/weather.csv"
start_position => beginning
}
}
이 설정은 Logstash가 어디에서 데이터를 찾아야 하는지, 그리고 파일의 처음부터 읽어 들이라는 의미입니다.
필터(filter) 섹션으로 데이터 가공하기
다음으로 필터가 필요합니다. Logstash는 기본적으로 다양한 필터 플러그인을 제공하며, 이 예제에서는 그중 몇 가지를 사용해 데이터를 파싱합니다. 아직 Logstash는 파일 안의 데이터에 대해 아무것도 모르기 때문에, 형식과 각 필드를 처리하는 세부 방식을 직접 지정해 주어야 합니다.
filter {
csv {
columns => [
"HumOut",
"TempIn",
"DewPoint",
"HumIn",
"WindDir",
"RainMonth",
"WindSpeed",
"RainDay",
"BatteryVolts",
"WindChill",
"Pressure",
"time",
"TempOut",
"WindSpeed10Min",
"RainRate"
]
separator => ","
remove_field => ["message"]
}
date {
match => ["time", "ISO8601"]
}
mutate {
convert => ["TempOut", "float"]
}
}
컬럼 정의는 이름만 봐도 직관적이지만, 조금 더 자세히 살펴보겠습니다. 첫째, 전체 행이 통째로 담기는 message 필드를 제거합니다. 특정 속성값을 검색하는 것이 목적이므로 이 필드는 필요 없습니다. 둘째, time 필드가 ISO8601 형식의 날짜임을 지정해 Elasticsearch가 이를 단순 문자열이 아닌 날짜 타입으로 인식하도록 합니다. 마지막으로 mutate 기능을 사용해 TempOut 값을 부동소수점 숫자로 변환합니다.
출력(output) 섹션으로 Elasticsearch에 저장하기
이제 다음 코드를 통해 데이터를 적재하고, Elasticsearch에 저장하면서 파싱 결과를 처리합니다.
output {
elasticsearch {
protocol => "https"
host => ["iad1-20999-0.es.objectrocket.com:20999"]
user => "erik"
password => "mysupersecretpassword"
action => "index"
index => "eriks_weather_index"
}
stdout { }
}
마지막으로 호스트와 포트, 인증 정보, 그리고 저장할 인덱스의 이름을 설정하면 됩니다.
실행하고 결과 확인하기
자, 이제 실행해 보겠습니다. 정상적으로 동작한다면 다음과 비슷한 출력이 나타납니다.
$ bin/logstash -f conf/logstash.conf -v
Logstash startup completed
제대로 되었는지 궁금하다면 Elasticsearch에 직접 물어보면 됩니다.
$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/_cat/indices?v'
health status index pri rep docs.count store.size pri.store.size
green open eriks_weather_index 5 1 294854 95.8mb 48.5mb
문서가 정상적으로 저장되었으니, 이번에는 하나를 조회해 보겠습니다.
$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/eriks_weather_index/_search?q=TempOut:>75&pretty&terminate_after=1'
이 요청은 TempOut 값이 75보다 큰(TempOut:>75) 문서를 찾고, 사람이 읽기 좋은 형식(pretty)으로 출력하며, 샤드당 최대 한 개의 결과만 반환(terminate_after=1)하라는 의미입니다. 실행하면 다음과 같은 응답을 받게 됩니다.
{
"took" : 4,
"timed_out" : false,
"terminated_early" : true,
"_shards" : {
"total" : 5,
"successful" : 5,
"failed" : 0
},
"hits" : {
"total" : 5,
"max_score" : 1.0,
"hits" : [ {
"_index" : "eriks_weather_index",
"_type" : "logs",
"_id" : "AU-yXZJIJb3HnhKvpdNC",
"_score" : 1.0,
"_source":{"@version":"1","@timestamp":"2015-06-22T10:24:23.000Z","host":"kibana","path":"/home/erik/weather.csv","HumOut":"86","TempIn":"79.7","DewPoint":"70.65179649787358","HumIn":"46","WindDir":"161","RainMonth":"2.7","WindSpeed":"0","RainDay":"0.36","BatteryVolts":"1.125","WindChill":"82.41464999999999","Pressure":"29.611","time":"2015-06-22T10:24:23Z","TempOut":75.1,"WindSpeed10Min":"0","RainRate":"0.0"}
} ]
}
}
성공입니다. Logstash는 흔해 빠진 갖가지 데이터를 Elasticsearch에서 자유롭게 다룰 수 있는 형태로 바꿔주는 훌륭한 만능 도구입니다. 지금 바로 여러분의 데이터에 적용해 보세요!
Rackspace에 대한 문의는 www.rackspace.com에서 Sales Chat을 클릭해 대화를 시작할 수 있으며, Feedback 탭을 통해 의견을 남기거나 질문할 수 있습니다.
The Rackspace Cloud 서비스 약관은 해당 페이지에서 확인하실 수 있습니다.