logstash配置文件

kafk2es.conf

input{
    kafka {
        bootstrap_servers => "10.10.10.13:9092, 10.10.10.14:9092, 10.10.10.15:9092, 10.10.10.16:9092, 10.10.10.17:9092"
        topics => ["dba-audit"]         #读取的topic名称
        group_id => "dba-audit"         #消费组名称，logstash集群消费kafka集群的身份标识，必须集群相同且唯一
        auto_offset_reset => "earliest" 
        codec => "json"
        consumer_threads => 10           #消费线程数，集群中所有logstash相加最好等于 topic 分区数
        client_id => "client-dba-audit" 
        max_poll_records => "150"
        max_poll_interval_ms => "600000"
        heartbeat_interval_ms => 2000
   }
}

filter {
	mutate { remove_field => "bid" }                  #去掉一些不必要的字段
	mutate { remove_field => "schema" }
	mutate { remove_field => "dbValType" }
	mutate { remove_field => "jdbcType" }
	mutate { remove_field => "entryType" }
	mutate { remove_field => "isDdl" }
	mutate { remove_field => "ddl" }
	mutate { remove_field => "pks" }
	mutate { add_field => { "hello" => "world" } }  #设置一个自定一字段"hello",对应的固定值为'world'
	ruby {
		code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*3600)"   # 将ES默认时间加上8小时的修改东八区时间
	}
	ruby {
		code => "event.set('@timestamp',event.get('timestamp'))"
	}
	mutate {
		remove_field => ["timestamp"]
	}
}


output {
    elasticsearch {
        index  =>  "dba-audit-%{+YYYY.MM.dd}"
        hosts => ["10.10.10.88:8089"]
        user => "elastic"
        password => "mypasswordxxxx"
        codec => json
    }
}