Nested objects from MySQL to ElasticSearch

Question

I am new to ES and trying to load data from MYSQL to Elasticsearch using logstash jdbc.

In my situation I want to use column values as field names, Please see new & hex in output data, I want 'id' values as field names.

Mysql data

cid    id       color      new     hex      create            modified
1      101     100 euro    abcd   #86c67c  5/5/2016 15:48   5/13/2016 14:15
1      102     100 euro    1234   #fdf8ff  5/5/2016 15:48   5/13/2016 14:15

output needed

{
  "_index": "colors_hexa",
  "_type": "colors",
  "_id": "1",
  "_version": 218,
  "found": true,
  "_source": {
    "cid": 1,
    "color": "100 euro",
    "new" : {
            "101": "abcd",
            "102": "1234",
        }
    "hex" : {
            "101": "#86c67c",
            "102": "#fdf8ff",
        }
    "created": "2016-05-05T10:18:51.000Z",
    "modified": "2016-05-13T08:45:30.000Z",
    "@version": "1",
    "@timestamp": "2016-05-14T01:30:00.059Z"
  }
}

Logstash config:

input {
 jdbc {
   jdbc_driver_library => "/etc/logstash/mysql/mysql-connector-java-5.1.39-bin.jar"
   jdbc_driver_class => "com.mysql.jdbc.Driver"
   jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test"
   jdbc_user => "root"
   jdbc_password => "*****"
   schedule => "* * * * *"

   statement => "select cid,id,color, new ,hexa_value ,created,modified from colors_hex_test order by cid"
   jdbc_paging_enabled => "true"
   jdbc_page_size => "50000"
}
}

   output {
    elasticsearch {
        index => "colors_hexa"
        document_type => "colors"
        document_id => "%{cid}"
        hosts => "localhost:9200"
    }
}

Can anyone please help with filter tag for this data, 'new' & 'hex' fields are the issue here. I m trying to convert two records to single document.

Answer 1

You're looking for the aggregate filter. One of their examples explicitly is for the JDBC use-case you're looking for here (see example 4).

Since the JDBC input is a scheduled action, you can set your aggregate filter to merge all events that arrive in a short period of time. Say, 10 seconds. All of the rows pulled in by the JDBC input will arrive very closely grouped, and should end up merged.

This filter is more complex than others, since you have to write ruby-code to handle the field-concatenation you're looking for. But it should be able to do it.