小编nan*_*boo的帖子

假设我的原文Map包含以下内容:

Map<String, Set<String>> original = Maps.newHashMap();
original.put("Scott", Sets.newHashSet("Apple", "Pear", "Banana");
original.put("Jack", Sets.newHashSet("Banana", "Apple", "Orange");

我想创建一个Map包含以下内容的反转:

  "Apple":  ["Scott", "Jack"]
  "Pear":   ["Scott"]
  "Banana": ["Scott", "Jack"]
  "Orange": ["Jack"]

我知道它可以用旧方式(Java 8之前的版本)完成,但是如何使用Java Stream API实现相同的功能呢？

Map<String, Set<String>> reversed = original.entrySet().stream().map(x -> ????).collect(??)

有张贴类似的问题在这里,但只适用于单值Map秒.

我正在使用 Debezium 来检测 MySql 源表中的更改。如何生成 Kafka 消息，使键为数字 ( Long) 值而不是 Json 对象？

我得到什么：

key: {"foo_id": 123} 
value: {"foo_id": 123, "bar": "blahblah", "baz": "meh......"}

我想要的是：

key: 123
value: {"foo_id": 123, "bar": "blahblah", "baz": "meh......"}

我的 FOO 表如下所示：

foo_id: INT
bar: VARCHAR 
baz: VARCHAR

请注意，我没有使用 avro，并且我已经尝试了以下几种组合（带和不带密钥变压器），但未能获得密钥Long。

"transforms": "unwrap,insertKey,extractKey",
"transforms.unwrap.type":"io.debezium.transforms.UnwrapFromEnvelope",
"transforms.unwrap.drop.tombstones":"false",
"transforms.insertKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
"transforms.insertKey.fields":"foo_id",
"transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.extractKey.field":"foo_id",        
"key.converter" : "org.apache.kafka.connect.converters.LongConverter",
"key.converter.schemas.enable": "false", 
"value.converter" : "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"

我不确定 ValueToKey 或 ExtractField 适用于（MySQL）源，但我低于 NPE。

Caused by: java.lang.NullPointerException
        at org.apache.kafka.connect.transforms.ValueToKey.applyWithSchema(ValueToKey.java:85)
        at org.apache.kafka.connect.transforms.ValueToKey.apply(ValueToKey.java:65)
        at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:44) …