首页 文章

如何在加入Kstreams后选择String格式数据中的空值

提问于
浏览 次
0

我已经在两个kstream上执行了连接操作，这些kstream由avro格式数据组成，然后我的键是Integer类型，值是string类型 . 输出是这样的：

[KSTREAM-MERGE-0000000016]: 1, {"id": 1, "name": "john", "age": 26}/{"id":1, "name": "d", "age": 67}
[KSTREAM-MERGE-0000000016]: 2, {"id": 2, "name": "jane", "age": 24}/{"id": 2, "name": "e", "age": 78}
[KSTREAM-MERGE-0000000016]: 3, {"id": 3, "name": "julia", "age": 25}/{"id": 3, "name": "h", "age": 12}
[KSTREAM-MERGE-0000000016]: 4, {"id": 4, "name": "jamie", "age": 22}/null
[KSTREAM-MERGE-0000000016]: 5, {"id": 5, "name": "jenny", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 6, {"id": 6, "name": "kishore", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 7, {"id": 7, "name": "purna", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 8, {"id": 8, "name": "xxx", "age": 10}/null
[KSTREAM-MERGE-0000000016]: 9, {"id": 9, "name": "yyy", "age": 10}/null
[KSTREAM-MERGE-0000000016]: 10, {"id": 10, "name": "zzz", "age": 10}/null

现在我想过滤掉最后有空的值 . 我的预期输出应该是：

[KSTREAM-MERGE-0000000016]: 4, {"id": 4, "name": "jamie", "age": 22}/null
[KSTREAM-MERGE-0000000016]: 5, {"id": 5, "name": "jenny", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 6, {"id": 6, "name": "kishore", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 7, {"id": 7, "name": "purna", "age": 27}/null
[KSTREAM-MERGE-0000000016]: 8, {"id": 8, "name": "xxx", "age": 10}/null
[KSTREAM-MERGE-0000000016]: 9, {"id": 9, "name": "yyy", "age": 10}/null
[KSTREAM-MERGE-0000000016]: 10, {"id": 10, "name": "zzz", "age": 10}/null
java apache-kafka avro apache-kafka-streams

1 回答

  • 0

    首先，删除null值

    KStream<Long, String> stream = streamsBuilder
        .stream("TOPIC1", Consumed.with(Long(), Serdes.String()))
        .filter((k, v) -> v != null);

    KStream<Long, String> stream2 = streamsBuilder
        .stream("TOPIC2", Consumed.with(Long(), Serdes.String()))
        .filter((k, v) -> v != null);

    stream.join(stream2, (value1, value2) -> value1 + "/" + value2, JoinWindows.of(1000));
    回复于

相关问题