需要删除kafka流中的null

Question

只需在我的 kafka-cosole-consumer 命令中删除 --property print.key=true 这个就可以在控制台上不显示 null ....但问题是它不仅仅是为了我正在使用的显示目的... .我的 kafka 流输出需要直接进入流反应器，然后从那里到 influxdb ....我的流反应器将接受

{"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"}

那么解决这个问题的方法是什么？请帮助我

Answer 1

您可以过滤这些消息：

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

或者您放置一个替代值来替换 null:

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