通过logstash导入csv数据到ElasticSearch

说明：logstash通过执行logstash.conf文件操作，其中logstash.conf文件包含（需配置信息）：导入数据文件位置，上次加载位置，转化格式。

1. 配置logstash.conf
   例如：需要导入电影相关数据

input {
  file {
    path => "YOUR_FULL_PATH_OF_movies.csv"//数据路径
    start_position => "beginning"
    sincedb_path => "/dev/123"   //历史位置存放文件，没有就新建
  }
}
filter {
  csv {
    separator => ","
    columns => ["id","content","genre"]
  }

  mutate {
    split => { "genre" => "|" }
    remove_field => ["path", "host","@timestamp","message"]
  }

  mutate {

    split => ["content", "("]
    add_field => { "title" => "%{[content][0]}"}
    add_field => { "year" => "%{[content][1]}"}
  }

  mutate {
    convert => {
      "year" => "integer"
    }
    strip => ["title"]
    remove_field => ["path", "host","@timestamp","message","content"]
  }

}
output {
   elasticsearch {
     hosts => "http://localhost:9200"
     index => "movies"
     document_id => "%{id}"
   }
  stdout {}
}

2. 在logstash的bin目录下执行conf文件
   命令如下：

 logstash -f YOUR_FULL_PATH_OF_logstash.conf 

结果如下：