ETL之Logstash - 抽取csv到elasticsear

1. 设置需要分析的csv 路径及文件名

> input {
> 
> # stdin { type => "earthquake" }
> 
> file {
> 
>     path => "/home/elastic/logstash-6.2.4/e.csv"
> 
>     start_position => "beginning"
> 
>     sincedb_path => "/dev/null"
> 
>   }
> 
> }

2. 使用csv 插件 分解具体fields, 并进行类型转换

> filter {
> 
>   csv {
> 
>     separator => ","
> 
>     columns => ["timestamp","latitude","longitude","depth","mag","magType","nst","gap","dmin","rms","source","event_id"]
> 
>     convert => {"latitude" => "float"}
> 
>     convert => {"longitude" => "float"}
> 
>     convert => {"depth" => "float"}
> 
>     convert => {"mag" => "float"}
> 
>     convert => {"dmin" => "float"}
> 
>     convert => {"rms" => "float"}
> 
>     convert => {"gap" => "float"}
> 
>   }

可以对部分字段做进一步的修改

> mutate {
> 
>     add_field => ["location", "%{latitude}, %{longitude}"]
> 
>     remove_field => ["latitude", "longitude"]
> 
>   }
> 
>   date {
> 
>     match => ["timestamp", "yyyy/MM/dd HH:mm:ss.SS", "ISO8601"]
> 
>     remove_field => ["timestamp"]
> 
>   }
> 
> }

3. 设定导出elasticsearch 地址， 加入配置了x-pack， 需要设置用户名密码

> output {
> 
> #  stdout { codec => rubydebug { metadata => true } }
> 
>   stdout { codec => dots }
> 
>   elasticsearch {
> 
>     hosts => ["10.2.28.8"]   
> 
> #    user => "elastic"
> 
> #    password => "changeme"
> 
>     index => "ncedc-earthquakes"
> 
>   }
> 
> }

Ref:

https://www.elastic.co/guide/en/logstash/current/plugins-filters-csv.html