版本：logstash 6.6.2
方法：用logstash过滤器中的json插件配置选项实现

首先介绍用到的参数
参数：add_field
作用：添加字段

参数：%{field}
作用：调用事件中的指定字段

参数：remove_field
作用：删除字段

官方插件资料：https://www.elastic.co/guide/en/logstash/6.6/plugins-filters-json.html
通过文档描述，json插件可以将事件中的json格式日志提取到事件的根中，根中的字段传递到elasticsearch后，可以做过滤条件，而且展示起来比较直观。

比如json日志格式为：

{
    "name":"root",
    "info":{
        "from":"host1",
        "path":"var/log/log.log"
    }
}

info字段的值还是一个json，现在需要将info中的数据提取到事件的根中，也就是变为如下格式。

{
    "name":"root",
    "from":"host1",
    "path":"var/log/log.log"
}

# logstash配置文件

[root@localhost logstash]# cat kafka.conf 
input {

	kafka {
		bootstrap_servers => ["192.168.10.74:9092"]
		client_id => "test1"
		group_id => "test1"
		auto_offset_reset => "latest"
		consumer_threads => 1
		decorate_events => false
		topics => ["test"]
		type => "fromk"
	}
 
}

# 将message字段的数据(json格式) "导入"
filter {
	json {
		# 输入的日志，都会进入message字段，使用source解析json。
		source => "message"
		# 添加列，便于下面再执行source
		add_field => { "javalog" => "%{info}" }
	}

	# 第二次解析json串
	json {
		source => "javalog"
		# 移除列，info字段和javalog字段不需转存到elasticsearch。
		remove_field => [ "info","javalog" ]
	}
}


output {

	elasticsearch {
		hosts => "192.168.10.74"
		index => "jar-log-%{+YYYY.MM.dd}"
	}

	stdout {
		codec => rubydebug
	}
}

效果：

kibana中展示效果