概述
记录自己在工作中将生产的数据按月保存在ES中（通过logstash采集kafka数据到ES），由于生产环境数据量比较庞大（一天的日志量大概在2500万条左右），为了后期减轻服务器压力，方便我们维护，所以需要对我们的日志进行处理，按月建立不同的ES索引库，能够查询最近6个月的日志，关闭前6个月不用的日志。
创建模板
如果用户每次新建一个索引的时候都需要手动创建mapping非常麻烦，es内部维护了template，template定义好了mapping，只要index的名称被template匹配到，那么该index的mapping就按照template中定义的mapping自动创建。而且template中定义了index的shard分片数量、replica副本数量等等属性。
模板样例
1{ 2 "order": 0, // 模板优先级 3 "template": "sample_info*", // 模板匹配的名称方式 4 "settings": {...}, // 索引设置 5 "mappings": {...}, // 索引中各字段的映射定义 6 "aliases": {...} // 索引的别名 7}
settings定义了索引的属性，包括分片数量、副本数量、写入flush时间间隔。
1"settings": { 2 "index": { 3 "refresh_interval": "10s",//每10秒刷新 4 "number_of_shards" : "5",//主分片数量 5 "number_of_replicas" : "2",//副本数量 6 "translog": { 7 "flush_threshold_size": "1gb",//内容容量到达1gb异步刷新 8 "sync_interval": "30s",//间隔30s异步刷新（设置后无法更改） 9 "durability": "async"//异步刷新 10 } 11 } 12 }
编写template，写入es
通过命令 put _template/[template名]{……}命令写入es
通过命令 get _template/[template名]查看是否写入template成功。如果template编写错误重新执行put命令覆盖即可
附 自己的template
1PUT /_template/template_access_log 2{ 3 "template": "poc*", 4 "settings": { 5 "index": { 6 "refresh_interval": "10s", 7 "number_of_shards" : "5", 8 "number_of_replicas" : "1", 9 "translog": { 10 "flush_threshold_size": "1gb", 11 "sync_interval": "30s", 12 "durability": "async" 13 } 14 } 15 }, 16 "mappings": { 17 "doc": { 18 "dynamic_templates": [ 19 { 20 "string_as_keyword": { 21 "match_mapping_type": "string", 22 "mapping": { 23 "type": "keyword" 24 } 25 } 26 } 27 ], 28 "properties": { 29 "TRANSAMOUNT": { 30 "type": "double" 31 }, 32 "consumeTime": { 33 "type": "integer" 34 }, 35 "transTime": { 36 "type": "date", 37 "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" 38 } 39 } 40 } 41 }, 42 "aliases" : { 43 "poc" : {} 44 } 45 }
logstash6.5.4配置文件修改
1input { 2 kafka{ 3 group_id=> "logstash_bkq_test001" 4 topics=> "test_bkq" 5 bootstrap_servers=> "192.168.8.21:9092" 6 key_deserializer_class=> "org.apache.kafka.common.serialization.IntegerDeserializer" 7 value_deserializer_class=> "org.apache.kafka.common.serialization.StringDeserializer" 8 auto_offset_reset=> "latest" 9 client_id=> "logstash_bkq_test001" 10 } 11} 12 13filter { 14 json { 15 source => "message" 16 remove_field => ["message"] 17 } 18} 19 20output { 21 elasticsearch { 22 hosts => ["192.168.8.23:9200","192.168.8.24:9200","192.168.8.25:9200"] 23 index => "poc%{+yyyyMMdd}" 24 user => "elastic" 25 password => "******" 26 } 27}
logstash往ES灌数据的时候按照index => "poc%{+yyyyMMdd}" 灌数据。