Logstash 从 JSON 导入 Elasticsearch

Question

测试数据

movies.json

{"movieId":1,"title":"Toy Story (1995)","genres":"Adventure|Animation|Children|Comedy|Fantasy","cast":[{"name":"Tom Hanks"},{"name":"Tim Allen"}]}
{"movieId":2,"title":"Jumanji (1995)","genres":"Adventure|Children|Fantasy","cast":[{"name":"Robin Williams"},{"name":"Bonnie Hunt"}]}
{"movieId":3,"title":"Grumpier Old Men (1995)","genres":"Comedy|Romance","cast":[{"name":"Jack Lemmon"},{"name":"Ann-Margret"}]}
{"movieId":4,"title":"Waiting to Exhale (1995)","genres":"Comedy|Drama|Romance","cast":[{"name":"Whitney Houston"},{"name":"Angela Bassett"}]}
{"movieId":5,"title":"Father of the Bride Part II (1995)","genres":"Comedy","cast":[{"name":"Steve Martin"},{"name":"Diane Keaton"}]}
{"movieId":100,"title":"Something Wrong","genres":"Ignore"}

Logstash 配置文件

logstash.conf

input {
  file {
    path => "/Users/jasonwu/WorkSpace/learn/Elasticsearch/movies.json"
    start_position => "beginning"
    sincedb_path => "/Users/jasonwu/.Trash/sincedb.trash"
  }
}

filter {
  json {
    source => "message"
  }

  if [moveId] == "100" or [genres] == "Ignore" {
    drop {}
  }

  split {
    field => "[cast]"
  }

  mutate {
    add_field => {
      "cast_name" => "%{[cast][name]}"
    }

    rename => {
      "movieId" => "movie_id"
      "genres" => "genre"
    }

    convert => {
      movie_id => "integer"
    }

    remove_field => ["cast", "path", "@version", "message", "@timestamp", "host"]
  }
}

output {
  stdout {}

  elasticsearch {
    hosts => ["localhost:9200"]
    index => "movies-json"
  }
}

执行 Logstash 导入 Elasticsearch

$ bin/logstash -f /Users/jasonwu/WorkSpace/learn/Elasticsearch/logstash.conf