如何在加入kstreams后选择字符串格式数据中的空值

atmip9wb 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(240)

我在两个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

现在我想过滤掉结尾有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 avro apache-kafka apache-kafka-streams

来源：https://stackoverflow.com/questions/50674193/how-to-select-null-values-in-string-format-data-after-joining-the-kstreams

1条答案

按热度按时间

zbsbpyhn1#

首先，删除空值

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));

赞(0）回复(0）举报 2021-06-07

我来回答

如何在加入kstreams后选择字符串格式数据中的空值

1条答案

相关问题

热门标签

最新问答