(logstash) Index only specific data in elasticsearch from kafka input

Question

I am using kafka as my input and put it in elasticsearch (output)

input {
    kafka {
        topics =>["maxwell"]
        codec => json
    }
}
filter {
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
        index => 'test_kafka'
        document_type => "%{table}"
        hosts => 'localhost:9200'
    }
}

When this runs, it outputs the following json

{
  "database": "my_db",
  "xid": 88935,
  "@timestamp": "2016-11-14T12:00:13.763Z",
  "data": {
    "contact_country_code": null,
    "contact_type_id": 1,
    "created": "2014-10-03 12:24:36",
    "modified_by": null,
    "modified": "2014-10-03 12:24:36",
    "contact_id": 1,
    "is_default": 0,
    "created_by": null,
    "contact_number": "1241222232"
  },
  "old": {
    "contact_number": "1241222"
  },
  "commit": true,
  "@version": "1",
  "type": "update",
  "table": "contact",
  "ts": 1479124813
}

My question is, how can I only extract the data key with dynamic document_type in elasticsearch to achieve the this one

{
  "_index": "test_kafka",
  "_type": "contact",
  "_id": "AVhitY804rvpX8qdVt9d",
  "_score": 1,
  "_source": {
    "contact_country_code": null,
    "contact_type_id": 1,
    "created": "2014-10-03 12:24:36",
    "modified_by": null,
    "modified": "2014-10-03 12:24:36",
    "contact_id": 1,
    "is_default": 0,
    "created_by": null,
    "contact_number": "1241222232"
  }
}

Val · Accepted Answer · 2016-11-15 09:46:27Z

1

You can add a ruby filter to massage your event like below. What it does is first save the table field inside the @metadata fields so you can reference it in your elasticsearch output. Then it deletes all fields except the data one. Then it copies all fields inside the data field at the root level and finally it deletes the data field.

input {
    kafka {
        topics =>["maxwell"]
        codec => json
    }
}
filter {
  mutate {
     add_field => { "[@metadata][type]" => "%{table}" }
  }
  ruby {
     code => "
        # Ruby code for Logstash 2.x
        event.to_hash.delete_if {|k, v| k != 'data'}
        event.to_hash.update(event['data'].to_hash)
        event.to_hash.delete_if {|k, v| k == 'data'}

        # Ruby code for Logstash 5.x
        event.to_hash.delete_if {|k, v| k != 'data'}            
        event.to_hash.update(event.get('data').to_hash)
        event.to_hash.delete_if {|k, v| k == 'data'}
     "
  }        
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => 'localhost:9200'
        index => 'test_kafka'
        document_type => "%{[@metadata][type]}"
    }
}

edited Nov 15, 2016 at 9:46

answered Nov 15, 2016 at 5:03

Val

218k14 gold badges377 silver badges384 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

scireon Over a year ago

The event didn't mutate, btw, I'm using logstash 5.0

scireon Over a year ago

I've also tried replacing the event with this code ``` event = event.get('data') ``` but didn't worked as expected

Val Over a year ago

Ok, indeed, that's ruby code for Logstash 2.x. Let me fix it for Logstash 5.

Collectives™ on Stack Overflow

(logstash) Index only specific data in elasticsearch from kafka input

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related