logstash如何使用动态的库名表名写入Doris

[xueliang787]

想在写入doris的时候按照上游的不同的kafka 的topic分不同的表，但是语法有问题，这块该如何解决，配置如下。例如

output {
doris {
http_hosts => ["http://localhost:8630"]
user => "root"
password => ""
db => "doris"
table => "%{topic}"
headers => {
"format" => "json"
"read_json_by_line" => "true"
"load_to_single_tablet" => "true"
}
mapping => {
"log_time" => "%{log_time}"
"collect_time" => "%{@timestamp}"
"host" => "%{[host][name]}"
"path" => "%{[log][file][path]}"
"type" => "%{type}"
"level" => "%{level}"
"thread" => "%{thread}"
"position" => "%{position}"
"message" => "%{message}"
}
log_request => true
}
}

上游input是kafka的很多个topic，中间filter做了简单解析，传入topic变量，下游的table /db