ELK同步kafka帶有key的Message

需求

kafka中的message帶有key，帶有相同key值的message后入kafka的意味着更新message，message值為null則意味着刪除message。
用logstash來同步kafka中這樣的數據入Elasticsearch，從而實現可以同步增刪改數據。

環境

1) logstash-6.5.4 
2) kafka中topic的message帶有key

解決(Logstash的配置如下）

input {
    kafka {
        bootstrap_servers=> "192.168.31.92:9092"
        group_id => "test_group"
        topics => ["test_topic"]
        type => "test_type"
        auto_offset_reset => earliest
        consumer_threads => 1
        decorate_events => true
        codec => plain {
            format => ""
        }
    }
}

filter{
    if ([message] == "") {
        mutate {
            add_field => { "@esaction" => "delete"}
        }
	mutate {
            remove_field => ["@version"]
        }
    } else {
        json {
            source => "message"
        }

        mutate {
            remove_field => ["@version","message"]
        }

        mutate {
            add_field => { "@esaction" => "index"}
        }

        date {
            match => ["updated","UNIX_MS"]
	    target => "@timestamp"
        }
    }
}

output {
    elasticsearch {
        hosts => ["192.168.21.80:9200"]
        index => "test_index"
        document_id => "%{[@metadata][kafka][key]}"
        action => "%{[@esaction]}"
        codec => "json"
    }
}