Kafka流中的空值需要删除

zaqlnxep  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(288)

只是移除 --property print.key=true 在我的kafka cosole consumer命令中,这个命令用于在控制台上不显示null….但问题是,它不仅仅用于我正在使用的显示目的….我从kafka streams的输出需要直接到stream reactor,然后从那里到influxdb。。。。我的React堆可以接受

{"timestamp":"2017-10-04T10:43:56.614451246+05:30","namespace":"/intel/psutil/load/load1","data":0.42,"unit":"Load/1M","version":0,"last_advertised_time":"2017-10-04T10:43:56.669248345+05:30"}
{"timestamp":"2017-10-04T10:43:56.61445546+05:30","namespace":"/intel/psutil/load/load5","data":0.44,"unit":"Load/5M","version":0,"last_advertised_time":"2017-10-04T10:43:56.669251145+05:30"}

只有这么多……它不能接受

null    {"timestamp":"2017-12-05T13:52:43.560588466+05:30","namespace":"/intel/procfs/meminfo/active_anon","data":4183900160,"unit":"","tags":{"plugin_running_on":"AELAB110"},"version":4,"last_advertised_time":"2017-12-05T13:52:43.560658038+05:30"}
null    {"timestamp":"2017-12-05T13:52:43.560618652+05:30","namespace":"/intel/procfs/meminfo/mem_free","data":49385746432,"unit":"","tags":{"plugin_running_on":"AELAB110"},"version":4,"last_advertised_time":"2017-12-05T13:52:43.560660098+05:30"}

那么解决这个问题的办法是什么呢?请帮帮我

b09cbbtk

b09cbbtk1#

您可以过滤这些消息:

stream.filter(...).to(...);

或者你把一个替代值 null :

stream.map(...).to(...);

相关问题