"问题：

从windows 通过flume传输到kafka的日志（GBK），然后再logstash 消费，用json 解析。有些日志解析报错：ctrl-code 1(SOH 文本头)。分析：在终端上用gbk编码查看会有 方框的字符，可以判断该字符就是json 无法识别的控制字符（SOH）。但是不知道kafka消费的时候显示成什么字符，如果知道这个（SOH）然后替换成，json 可以解析的字符就可以了。

解决办法：

在日志里发现每一个（SOH）处，都显示为\u0001,这是16进制的ascll code 1。和控制字符的code1也是对应的，断定（SOH)被解析成\u0001这个字符。input kafka中之前都是直接codec => json{charset =>[""GBK""]},相当于在消费的时候先进行json 解析，这样很无力，一直报错。

换个思路：

1，先无格式plain消费，然后把SOH(\u0001) 2，替换成空格（因为soh无意义，替换成空格不影响源信息）gsub =>["message","\u0001"," "]，3，再用 json {source => ""message""} 进行解析，这样就可以了。--------------------------配置信息---------------------------------input { kafka { bootstrap_servers => ""ZBSZ1-LOG-KFK01:9092,ZBSZ1-LOG-KFK02:9092,ZBSZ1-LOG-KFK03:9092""group_id => ""es-rzrqbp02"" topics_pattern => ""rzrqbp-C010001""value_deserializer_class => ""org.apache.kafka.common.serialization.ByteArrayDeserializer"" #源字节编码转换器，因为一直以GBK编码传输codec => plain{charset => [""GBK""]}#codec => json}} filter { mutate { convert => { ""[indicator][usedtime]"" => ""integer"" }gsub =>[""message"",""\u0001"","" ""]}json { source => ""message""}date { match => [""[time_stamp]"",""UNIX_MS""]target => ""@timestamp""}

}

output {

elasticsearch { hosts => [""ZBSZ1-LOG-ES01:9200"", ""ZBSZ1-LOG-ES02:9200"", ""ZBSZ1-LOG-ES03:9200""]index => ""app-rzrqbp-%{+YYYY.MM.dd}""document_id => ""%{[indicator][msgid]}""}

}

"