ELK生态：Logstash增量读取txt文件数据，导入到Elasticsearch

蔚落 2022-01-14 09:15 1137阅读 0赞

### 简介 ###

1.  ELK生态之Logstash导入数据到Elasticsearch；
2.  数据源：txt格式文件，内容为json或json数组；
3.  Elasticsearch和Logstash版本：5.6.1；
4.  前提环境：Elasticsearch单机或集群；Logstash客户端；

### 实践 ###

*  txt文件内容：

[{"name":"sixmonth","sex":"男"},{"name":"xiaoming","sex":"男"}]
    [{"name":"xiaoxie","sex":"男"},{"name":"xiaodou","sex":"男"}]

*  logstash.conf配置：

#1.读取数据txt文件，数据输入阶段
    input {
        file{
            #设置txt文件路径，多个文件路径可设置成数组[]，模糊匹配用*
            #指定单一文件
            #path => "/data/es/logstash-5.6.1/files/test.txt"
            #指定数组文件
            #path => ["/data/es/logstash-5.6.1/files/test-1.txt","/data/es/logstash-5.6.1/files/test-2.txt"]
            #指定同级目录模糊匹配
            #path => "/data/es/logstash-5.6.1/files/test*.txt"
            #指定多级目录模糊匹配
            path => "/data/es/logstash-5.6.1/files/**/test*.txt"
            
            #设置logstash开始读取文件内容位置，begining为从头开始，end为只读取最新数据
            start_position => beginning
            
            #设置编码
            codec => json {charset => "UTF-8"}
            
            #设置输入输出标识
            type => "txt_index"
        }
     }  
      
    #2.过滤格式化数据阶段
    filter {
       
        mutate{
            #删除无效的字段
            remove_field => ["@version","message","host","path"]
        }
        
        #新增timestamp字段，将@timestamp时间增加8小时
        ruby { code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)" }
        
    }
    
    #3.数据输出到ES阶段
    output {
    
        #日志输出格式，json_lines;rubydebug等
        stdout {
            codec => rubydebug
        }
        
        #输出到es
        if[type] == "txt_index"{
        
            #无法解析的json不记录到elasticsearch中
            if "_jsonparsefailure" not in [tags] {
                elasticsearch {
                    #es地址ip端口
                    hosts => "127.0.0.1:9200"
                    #索引
                    index => "txt_index"
                    #类型
                    document_type => "txt_index"
                    #覆盖模板，不需要可注释掉，通用模板下载：https://download.csdn.net/download/alan_liuyue/11241484
                    template_overwrite=>true
                    template=>"/data/es/logstash-5.6.1/template/logstash.json"            
                    
                }
            }
        }
        
    }

*   补充file-input字段说明：

codec => #可选项，默认是plain，可设置其他编码方式;
    
    discover_interval => #可选项，logstash多久检查一下path下有新文件，默认15s;
    
    exclude => #可选项，排除path下不想监听的文件;
    
    sincedb_path => #可选项，记录文件以及文件读取信息位置的数据文件;
    
    sincedb_write_interval => #可选项，logstash多久写一次sincedb文件，默认15s;
    
    stat_interval => #可选项，logstash多久检查一次被监听文件的变化，默认1s;
    
    start_position => #可选项，表示从哪个位置读取文件数据，初次导入为：beginning，最新数据为：end
    
    path => #必选项，配置文件路径，可定义多个，也可模糊匹配;
    
    tags => #可选项，在数据处理过程中，由具体的插件来添加或者删除的标记;
    
    type => #可选项，当有多个file的时候，可用于一对一匹配输入或者输出;

### 注意事项 ###

1.  logstash启动之后，进程会一直处于运行状态，若txt文件被修改，程序会自动监听，导入新数据；
2.  若需要重新导入数据，则需要删除logstash安装目录下 \\plugins\\inputs\\file 下的文件(该文件属于隐藏文件)，所以直接删除该目录即可，之后目录会重新生成；

### 总结 ###

1.  Logstash读取csv文件内容导入Elasticsearch，在数据源采集方面特别普遍，采集配置方式也特别简洁，程序猿有兴趣可继续深入挖掘；
2.  实践是检验认识真理性的唯一标准，自己动手丰衣足食~