java使用KStream过滤掉阈值以外的值

Question

我想在Kafka中使用JavaKStream来过滤掉所有超过某个值的值。值交换为JSON，例如：

ConsumerRecord(topic=u'test', partition=0, offset=1109, timestamp=1528110096230L, timestamp_type=0, key=None, value='{"device":"Internal","sensor":"Phone Microphone","value":"72.1"}', checksum=None, serialized_key_size=-1, serialized_value_size=64)

我想过滤掉低于20.0的值（在上面的例子中，值是72.1，没有问题）

public class WordCountExample {

@SuppressWarnings("deprecation")
public static void main(String[] args) throws Exception{

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "Filter");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "andrewnetwork.ddns.net:9095");
    props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");


    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> source = builder.stream("test");

    source =  source
    .filterNot((k,v) -> {
        if(isParsableAsDouble(v) && Double.parseDouble(v) <= 50.0)
            return true;
        else return false;
    });

    source.to("mem");

过滤没有发生，我不知道为什么。有什么想法吗

Answer 1

通过将isParsableAsDouble(v)放在filterNot中，您过滤掉了所有内容，因为JSON不能作为双精度函数进行分析。我相信您误解了Kafka值和JSON中的value字段，后者不是自动提取的

您需要一个JSON反序列化程序For example

    final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
    final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
    final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);


    KStreamBuilder builder = new KStreamBuilder();
    Properties props = new Properties();
    // load props 

    KStream<Bytes, JsonNode> source = builder.stream(Serdes.BytesSerde(), jsonSerde, "test")
        .filter((k, v) -> {
            return v.get("value").asDouble() > 20.0;
         });