**日志存储：**1.5.4
**java:**1.8.0\u 102版
配置：

input {
        kafka {
        type => "steps"
                zk_connect => "kafka.server:2181"
                topic_id => "events"
                codec => "json"
        }
}

filter {
    if [type] == "steps" {
        if [event_name] == "Event_submitted" {
            aggregate {
                task_id => "%{step_id}"
                code => "map['step_duration'] = 0"
                map_action => "create"
                timeout => 3600
            }
        }
        if [event_name] == "Event_process_end" {
            aggregate {
                task_id => "%{step_id}"
                code => "event['step_duration'] = map['step_duration']"
                map_action => "update"
                end_of_task => true
                timeout => 3600
            }
        }
    }
}

output {
    file {
        type => "steps"
        path => "/opt/test.log"
    }
}

Kafka事件：

{"owner":"user","project_name":"NULL","step_id":"36777","component":"Deployment","stage_name":"stage1","branch_name":"master","additional_parameters":"NULL","received_record_ts":"1484749497752","event_name":"Event_submitted","timestamp":"20170118142457","event_group":"pipeline_events","@version":"1","@timestamp":"2017-01-18T14:25:00.818Z","type":"steps"}
{"owner":"user","project_name":"NULL","step_id":"36777","component":"Deployment","stage_name":"stage1","branch_name":"master","additional_parameters":"result:success","received_record_ts":"1484749497800","event_name":"Event_process_end","timestamp":"20170118142457","event_group":"pipeline_events","@version":"1","@timestamp":"2017-01-18T14:25:00.823Z","type":"steps"}

预期结果：

{"message" => "step_name:Events_processing,start_time:TIME,end_time:TIME,step_id:%{step_id},component:%{component},owner:%{owner},stage_name:%{stage_name},project:%{project_name},branch_name:%{branch_name},step_duration=%{step_duration}}

如何实现这一目标？通过这种配置，在输出中，我得到了完整的事件，而不是仅步骤