カフカストリームのヌルを削除する必要があります

Question

--property print.key=trueこれを私のkafka-cosole-consumerコマンドで削除すると、コンソールにはnullが表示されません....しかし、それは私が使用している表示目的だけではありません。 ...カフカからの私の出力は、それが

を受け入れることはできません.....私のストリームの原子炉は非常にだけ

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

これを受け入れるだろう.... influxdbし、そこから、その後、流れ反応器に直接移動する必要がありますストリーム

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

これはどのような解決策ですか？ plzz私

Answer 1

を助けるあなたはそれらのメッセージをフィルタリングすることができ、次のいずれか

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

それともnullを置き換えるサロゲート値を置く：

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