Elasticsearch 5의 가장 매력적인 신규 기능 중 하나는 바로 인제스트 노드(ingest node)입니다. 이 기능은 Logstash 스타일의 데이터 가공 기능을 Elasticsearch 클러스터 내부에 추가해 주기 때문에, 별도의 서비스나 인프라 없이도 데이터가 색인되기 전에 변환 작업을 수행할 수 있습니다. 예전에 Logstash로 CSV 파일을 파싱하는 방법에 관한 블로그 포스트를 올린 적이 있는데, 이번에는 비교 차원에서 동일한 작업을 인제스트 파이프라인으로 구현해 보겠습니다.
이 글에서는 Filebeat를 사용해 데이터를 인제스트 파이프라인으로 전송하고, 색인한 뒤, Kibana로 시각화하는 전체 과정을 예제와 함께 살펴봅니다.
데이터 살펴보기
무료로 활용할 수 있는 좋은 데이터 소스가 많지만, ObjectRocket의 직원 대부분이 텍사스주 오스틴에 거주하고 있어서 data.austintexas.gov의 데이터를 사용해 보겠습니다. 그중 음식점 위생 검사 데이터셋은 규모도 적당하고 실무에 가까운 예제를 만들기에 충분한 정보를 담고 있어 적합합니다.
아래는 해당 데이터셋의 몇 줄 발췌본입니다. 데이터 구조를 짐작할 수 있습니다:
Restaurant Name,Zip Code,Inspection Date,Score,Address,Facility ID,Process Description
Westminster Manor,78731,07/21/2015,96,"4100 JACKSON AVE
AUSTIN, TX 78731
(30.314499, -97.755166)",2800365,Routine Inspection
Wieland Elementary,78660,10/02/2014,100,"900 TUDOR HOUSE RD
AUSTIN, TX 78660
(30.422862, -97.640183)",10051637,Routine Inspection
앗, 문제가 생겼습니다. 하나의 항목이 한 줄로 깔끔하게 정리된 형태가 아니네요. 하지만 걱정하지 마세요. 곧 보시겠지만, Filebeat에는 여러 줄에 걸친 항목(multiline)을 처리하고 데이터 속에 숨어 있는 개행 문자를 우회하는 기능이 기본적으로 내장되어 있습니다.
참고로, 저는 원래 장애물 없는 간단한 예제를 만들 계획이었지만, 오히려 Elastic Stack이 제공하는 다양한 도구들을 어떻게 활용하면 이런 상황을 해결할 수 있는지 보여드리는 것이 더 흥미로울 것 같아 이렇게 진행했습니다.
Filebeat 설정하기
첫 번째 단계는 Filebeat가 Elasticsearch 클러스터로 데이터를 전송할 수 있도록 준비하는 것입니다. Filebeat를 다운로드(Elasticsearch 클러스터와 같은 버전을 사용하는 것이 좋습니다)하고 압축을 해제했다면, 함께 제공되는 filebeat.yml 설정 파일을 통해 아주 간단히 구성할 수 있습니다. 이번 시나리오에서 사용한 설정은 다음과 같습니다.
filebeat.prospectors:
- input_type: log
paths:
- /Path/To/logs/*.csv
# 컬럼 헤더가 있는 첫 번째 줄은 제외
exclude_lines: ["^Restaurant Name,"]
# 마지막 두 컬럼을 항목의 끝으로 식별하고, 앞선 줄들을 그 앞에 붙임
multiline.pattern: ',\d+,[^\"]+$'
multiline.negate: true
multiline.match: before
#================================ Outputs =====================================
output.elasticsearch:
# 연결할 호스트 배열
hosts: ["https://dfw-xxxxx-0.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-1.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-2.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-3.es.objectrocket.com:xxxxx"]
pipeline: "inspectioncsvs"
# 선택 사항: 프로토콜 및 기본 인증 정보
username: "esuser"
password: "supersecretpassword"
설정 자체는 상당히 직관적입니다. 입력 파일을 어디서 어떻게 수집할지 지정하는 섹션과, 데이터를 어디로 전송할지 지정하는 섹션으로 나뉩니다. 여기서 특별히 짚고 넘어갈 부분은 multiline 설정과 Elasticsearch 출력 설정 두 가지입니다.
이 데이터셋은 형식이 엄격하지 않습니다. 큰따옴표 사용이 일관되지 않고 개행 문자도 곳곳에 섞여 있죠. 따라서 가장 좋은 방법은 항목의 끝을 찾는 것이었습니다. 항목의 끝은 숫자 ID 뒤에 큰따옴표나 개행 없이 거의 변형 없이 오는 검사 유형이 붙는 패턴으로 구성됩니다. Filebeat는 패턴에 매칭되지 않는 줄들을 큐에 쌓아 두었다가, 패턴에 일치하는 마지막 줄 앞에 붙여 넣습니다. 만약 여러분의 데이터가 더 깔끔해서 한 줄에 한 항목씩 들어 있다면 multiline 설정은 거의 무시해도 됩니다.
Elasticsearch 출력 섹션을 보면 표준 Elasticsearch 설정에 pipeline: 디렉티브로 사용할 파이프라인 이름만 추가된 형태입니다. ObjectRocket 서비스를 사용 중이라면 UI의 "Connect" 탭에서 출력 스니펫을 복사하면 됩니다. 호스트 정보가 이미 채워져 있으므로 pipeline 라인만 추가하고 사용자명과 비밀번호를 입력하면 됩니다. 또한 아직 하지 않았다면, 클러스터 ACL에 자신의 시스템 IP를 반드시 추가해야 한다는 점도 잊지 마세요.
인제스트 파이프라인 생성하기
입력 데이터와 Filebeat가 준비되었으니, 이제 인제스트 파이프라인을 만들고 다듬을 차례입니다. 파이프라인이 수행해야 할 주요 작업은 다음과 같습니다:
- CSV 콘텐츠를 올바른 필드로 분리
- 검사 점수(Score)를 정수형으로 변환
@timestamp필드 설정- 그 외 데이터 형식 정리
다음은 이 모든 작업을 처리할 수 있는 파이프라인입니다:
PUT _ingest/pipeline/inspectioncsvs
{
"description" : "Convert Restaurant inspections csv data to indexed data",
"processors" : [
{
"grok": {
"field": "message",
"patterns": ["%{REST_NAME:RestaurantName},%{REST_ZIP:ZipCode},%{MONTHNUM2:InspectionMonth}/%{MONTHDAY:InspectionDay}/%{YEAR:InspectionYear},%{NUMBER:Score},\\"%{DATA:StreetAddress}\n%{DATA:City},?\\s+%{WORD:State}\\s*%{NUMBER:ZipCode2}\\s*\n\\(?%{DATA:Location}\\)?\\",%{NUMBER:FacilityID},%{DATA:InspectionType}$"],
"pattern_definitions": {
"REST_NAME": "%{DATA}|%{QUOTEDSTRING}",
"REST_ZIP": "%{QUOTEDSTRING}|%{NUMBER}"
}
}
},
{
"grok": {
"field": "ZipCode",
"patterns": [".*%{ZIP:ZipCode}\\"?$"],
"pattern_definitions": {
"ZIP": "\\d{5}"
}
}
},
{
"convert": {
"field" : "Score",
"type": "integer"
}
},
{
"set": {
"field" : "@timestamp",
"value" : "//"
}
},
{
"date" : {
"field" : "@timestamp",
"formats" : ["yyyy/MM/dd"]
}
}
],
"on_failure" : [
{
"set" : {
"field" : "error",
"value" : " - Error processing message - "
}
}
]
}
Logstash와 달리, 인제스트 파이프라인은 이 글을 작성하는 시점 기준으로 csv 전용 프로세서/플러그인이 없기 때문에 CSV 변환을 직접 구현해야 합니다. 저는 각 행의 컬럼 수가 많지 않아 grok 프로세서로 주요 작업을 처리했습니다. 컬럼이 훨씬 많은 데이터라면 grok 프로세서가 다루기 어려워질 수 있는데, 이 경우 split 프로세서와 painless 스크립트를 조합해 좀 더 반복적인 방식으로 라인을 처리하는 방법도 고려해 볼 수 있습니다. 두 번째 grok 프로세서를 눈치채셨을 텐데, 이것은 이 데이터셋에서 우편번호가 두 가지 서로 다른 방식으로 입력되어 있어 이를 정리하기 위한 것입니다.
디버깅을 위해 범용 on_failure 섹션을 포함했습니다. 이 섹션은 모든 에러를 잡아내어 어떤 종류의 프로세서가 실패했는지, 그리고 파이프라인을 깨뜨린 메시지가 무엇인지 출력해 줍니다. 덕분에 디버깅이 훨씬 쉬워집니다. error 필드가 설정된 문서를 인덱스에서 조회한 뒤, simulate API로 디버깅하면 됩니다. 바로 이어서 살펴보겠습니다.
파이프라인 테스트하기
인제스트 파이프라인 구성이 끝났으니, 이제 simulate API로 테스트해 보겠습니다. 먼저 샘플 문서가 필요합니다. 방법은 두 가지가 있습니다. pipeline 설정 없이 Filebeat를 실행한 후 Elasticsearch에서 처리되지 않은 문서를 가져오거나, 아니면 Elasticsearch 출력 섹션을 주석 처리하고 yml 파일에 아래 내용을 추가해 콘솔 출력을 활성화한 상태로 Filebeat를 실행하는 방법입니다:
output.console:
pretty: true
다음은 제 환경에서 가져온 샘플 문서입니다:
POST _ingest/pipeline/inspectioncsvs/_simulate
{
"docs" : [
{
"_index": "inspections",
"_type": "log",
"_id": "AVpsUYR_du9kwoEnKsSA",
"_score": 1,
"_source": {
"@timestamp": "2017-03-31T18:22:25.981Z",
"beat": {
"hostname": "systemx",
"name": "RestReviews",
"version": "5.1.1"
},
"input_type": "log",
"message": "Wieland Elementary,78660,10/02/2014,100,\\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\\",10051637,Routine Inspection",
"offset": 2109798,
"source": "/Path/to/my/logs/Restaurant_Inspection_Scores.csv",
"tags": [
"debug",
"reviews"
],
"type": "log"
}
}
]
}
그리고 응답 결과입니다(우리가 설정하려던 필드들만 추려서 보여드립니다):
{
"docs": [
{
"doc": {
"_id": "AVpsUYR_du9kwoEnKsSA",
"_type": "log",
"_index": "inspections",
"_source": {
"InspectionType": "Routine Inspection",
"ZipCode": "78660",
"InspectionMonth": "10",
"City": "AUSTIN",
"message": "Wieland Elementary,78660,10/02/2014,100,\\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\\",10051637,Routine Inspection",
"RestaurantName": "Wieland Elementary",
"FacilityID": "10051637",
"Score": 100,
"StreetAddress": "900 TUDOR HOUSE RD",
"State": "TX",
"InspectionDay": "02",
"InspectionYear": "2014",
"ZipCode2": "78660",
"Location": "30.422862, -97.640183"
},
"_ingest": {
"timestamp": "2017-03-31T20:36:59.574+0000"
}
}
}
]
}
파이프라인이 성공적으로 실행되었고, 무엇보다 중요한 것은 모든 데이터가 알맞은 위치에 들어갔다는 점입니다.
Filebeat 실행하기
Filebeat를 실행하기 전에 마지막으로 한 가지 작업을 더 하겠습니다. 인제스트 파이프라인에 익숙해지는 것만이 목표라면 이 단계는 완전히 선택 사항입니다. 하지만 grok 프로세서에서 설정한 Location 필드를 Geo-point로 활용하고 싶다면 filebeat.template.json 파일에 매핑을 추가해야 합니다. properties 섹션에 다음 내용을 추가하세요:
"Location": {
"type": "geo_point"
},
이제 준비가 끝났으니, ./filebeat -e -c filebeat.yml -d "elasticsearch" 명령으로 Filebeat를 실행하면 됩니다.
데이터 활용하기
GET /filebeat-*/_count
{}
{
"count": 25081,
"_shards": {
"total": 5,
"successful": 5,
"failed": 0
}
}
좋은 신호입니다! 에러가 있는지도 확인해 보겠습니다:
GET /filebeat-*/_search
{
"query": {
"exists" : { "field" : "error" }
}
}
{
"took": 1,
"timed_out": false,
"_shards": {
"total": 5,
"successful": 5,
"failed": 0
},
"hits": {
"total": 0,
"max_score": null,
"hits": []
}
}
역시 좋은 신호입니다!
이제 Kibana에서 데이터를 시각화할 모든 준비가 끝났습니다. Kibana 대시보드 생성 과정은 다음 기회에 자세히 다루겠지만, 날짜, 음식점 이름, 점수, 위치 정보가 모두 있으니 멋진 시각화를 만들 자유도는 충분히 확보되어 있습니다.
마무리하며
결국 인제스트 파이프라인은 꽤 강력한 기능으로, 변환 작업을 손쉽게 처리할 수 있습니다. 모든 데이터 처리를 Elasticsearch로 옮기고, 호스트에는 경량 Beats만 설치하면 되므로 파이프라인 어딘가에 Logstash를 둘 필요가 없어집니다. 다만 인제스트 노드는 여전히 Logstash에 비해 몇 가지 부족한 점이 있습니다. 예를 들어, 인제스트 파이프라인에서 사용할 수 있는 프로세서의 종류가 아직 제한적이라 CSV 파싱 같은 단순한 작업조차 Logstash만큼 쉽지 않습니다. 다행히 Elasticsearch 팀은 새로운 프로세서를 꾸준히 출시하고 있는 것으로 보이며, 두 제품 간의 격차가 점점 줄어들기를 기대해 봅니다.