Logstash作為一個數據處理管道，提供了豐富的插件，能夠從不同數據源獲取用戶數據，進行處理后發送給各種各樣的后台。這中間，最關鍵的就是要對數據的類型就行定義或映射。
本文討論的 ELK 版本為 5.5.1。
為什么要定義數據
Elastisearch不僅是一個強大的全文檢索引擎，它還能夠對一些數據類型進行實時的統計運算，相關的結果可以通過Kibana的圖表展現出來。如果數據類型沒有正確的定義，那么Elasticsearch就無法進行運算了，因此，雖然數據類型的定義需要花一點時間，但你會收到意想不到的效果。
JSON、字符串和數字
所有送往Elasticsearch的數據都要求是JSON格式，Logstash所做的就是如何將你的數據轉換為JSON格式。ElasticSearch會幫我們自動的判斷傳入的數據類型，這么做當然極大的方便了用戶，但也帶來了一些問題。
Elastic中的一些數據類型： text、keyword、date、long、double、boolean、ip、object、nested、geo_point等。不同的類型有不同的用途，如果你需要全文檢索，那應該使用text類型，如果你需要統計匯總那應該選擇數據或者keyword類型。感謝動態映射 Dynamic Mapping 的存在，在向ES送數的時候我們不需要事先定義映射關系，ES會對新增的字段自動進行映射。但是你比Elasticsearch更加熟悉你的數據，因此可能需要自己進行顯示定義 Explicit Mapping 映射關系。例如IP字段，默認是解析成字符串，如果映射為IP類型，我們就可以在后續的查詢中按照IP段進行查詢，對工作是很有幫助的。我們可以在創建索引時定義，也可以在索引創建后定義映射關系。
對於已經存在的數據，無法更新映射關系。更新映射關系意味着我們必須重建索引。
先來看下面這個JSON文檔。
{
"@timestamp": "2017-08-11T20:11:45.000Z",
"@version": "1",
"count": 2048,
"average": 1523.33,
"host": "elasticsearch.com"
}
這里有五個字段：@timestamp，@version，count，average，host。其中 @timestamp 和 host 是字符串，count、average 是數字，@version比較特殊，它的值是數字，但是因為放在雙引號中，所以作為字符串來對待。
嘗試把數據送入到 Elasticsearch 中，首先創建一個測試的索引：
將數據存入索引
查看數據映射的情況
根據結果可知，在沒有明確定義數據類型的情況下，Elasticsearch會自動判斷數據的類型，因此 @timestamp、@version、host都被映射為 text ，average、count 被映射為數字。
在Logstash中定義數據類型映射
Logstash提供了 grok 和 mutate 兩個插件來進行數值數據的轉換。
grok
grok 目前是解析非結構化的日志數據最好的插件。特別適合處理syslog、apache或其他web服務器、mysql等為了閱讀而輸出的信息類日志。
grok 的基本用法如下：%{SYNTAX:SEMANTIC}，SYNTAX是grok提供的樣式Pattern的名稱，grok提供了120多種Pattern，SEMANTIC是你給匹配內容的名稱（標志符）。因為grok實際上是正則匹配，因此任何輸出都默認轉換為字符類型，如果你需要數據類型的轉換，則使用下面這種格式
%{NUMBER:SEMANTIC:int}
目前，類型轉換僅支持 int 和 float 兩種類型。
如果將帶小數的數字轉換為 int 類型，會將小數后的數字丟棄。
mutate
mutate 為用戶提供了處理Logstash event數據的多種手段。允許我們移除字段、重命名字段、替換字段、修改字段等操作。
filter {
mutate {
convert => { "num" => "integer" }
}
}
使用模版進行字段映射
Elasticsearch中通過模板來存放索引字段的映射關系，logstash可以在配置文件中指定模板文件來實現自定義映射關系。
查詢 Elasticsearch 中的模板，系統自帶了 logstash-* 的模板。
我們用實際的例子來看一下映射和模板是如何起作用的。
1、首先創建一個 logstash 配置文件，通過 filebeat 讀取 combined 格式的 apache 訪問日志。
配置文件名為 filebeat.conf 位於 logstash 文件夾內。filebeat的配置比較簡單，可以參考我的上一篇文章 Filebeat+Logstash+ElasticSearch+Kibana搭建Apache訪問日志解析平台
input {
beats {
port => "5043"
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
target => ["datetime"]
}
geoip {
source => "clientip"
}
}
output {
elasticsearch {
hosts => "localhost:9200"
index => "my_index"
#template => "/data1/cloud/logstash-5.5.1/filebeat-template.json"
#template_name => "my_index"
#template_overwrite => true
}
stdout { codec => rubydebug }
}
第一次數據導入的時候，我們先不使用模板，看看 es 如何默認映射數據，啟動elk環境，進行數據導入。
[maserati@iZ627x15h6pZ logstash-5.5.1]$ ../elasticsearch-5.5.1/bin/elasticsearch
[maserati@iZ627x15h6pZ logstash-5.5.1]$ ./bin/logstash -f filebeat.conf
[maserati@iZ627x15h6pZ filebeat-5.5.1-linux-x86_64]$ sudo ./filebeat -e -c filebeat.yml -d "publish"
數據導入完成后，看一下索引的情況
因為從log導入的數據，所以mapping中給映射規則起名為log，對應的是 document_type，可以看到clientip和 geoip.location 分別解析成了文本和數值。其他大部分內容都映射為 text 。這種不需要我們定義映射規則的處理方式非常方便，但有時候我們更需要精確的映射。
看一下ES映射模板，只有logstash命名的模板，因為名稱不匹配，所以沒有應用這里的映射規則。
這里可以注意到模板文件和索引中的映射關系稍有不同，沒關系，我們把 my_index 的映射關系拷貝下來，存為 filebeat-template.json ，這里貼一下一個刪減版的 模板文件。
{
"template": "my_index",
"order": 1,
"settings": {
"index.refresh_interval" : "5s"
},
"mappings": {
"_default_": {
"properties": {
"clientip" : { "type":"ip" },
"geoip": {
"properties": {
"location": { "type":"geo_point" }
}
}
}
}
}
}
我們可以通過命令行收工把模板上傳到 elasticsearch ，也可以通過 logstash 配置文件指定。
curl -XPUT http://localhost:9200/_template/my_index_template?pretty -d @filebeat-template.json
我的例子中，我們只需要把 filebeat.conf 中的注釋打開即可。然后刪除索引，我們對索引重建一下。
看一下索引，可以看到模板中定義的規則已經在里面了。
看一下索引字段，看到 clientip 已經定義成 ip 類型了。
同樣，geoip.location映射成 geo_point 類型。
這樣我們就可以做訪客地圖了。
這時，再看一下 template 的情況。
可以看到，除了默認的模板，新增了一個我們定義的 my_index 模板。后續還可以對模板進行修改，但是注意只能增加或者刪除，無法對已經映射的字段進行更新。
參考資料：
1、Using Logstash to help create an Elasticsearch mapping template
2、Using grok and mutate to type your data
3、Elasticsearch Mapping
4、Grok Filter Plugin
5、Mutate Filter Plugin
6、用logstash導入ES且自定義mapping時踩的坑