如何使用kafka流将选择字段从键和值从Topic1推送到Topic2

Question

我正在使用KStream，并且只想从键中提取和推送字段的一个子集，并将值从Topic1推送到Topic2。

对于ex，如果我在邮件中的值包含id、name、address、phoneno等字段.我只想把身份证，名字和地址推到一个新的话题上。这类似于replacefield.whitelist转换，但我想尝试使用KStream。

我可以在单个字段中这样做，但不确定如何为多个字段准备新值(结果是一个GenericRecord)。

Answer 1

您可以使用KStream.map函数访问键和值，然后返回一个新的键值对。

stream.map((k, v) -> {
    GenericRecord r = new GenericData.Record();  // needs a schema 
    r.put("name", v.getName());  // for example 
    return new KeyValue(k, r);
}).to("topic2")

如果不使用Avro Serde或其他泛型类型，则还需要定义自己的serde类。