如何使用 kafka 流加入主题答案

【问题标题】：How to join topics with kafka stream如何使用 kafka 流加入主题
【发布时间】：2020-05-01 17:14:19
【问题描述】：

卡夫卡流我正在尝试流式传输，但我遇到了一些问题，它不起作用。首先，我有三个连接器，但我不能使用自己的密钥。我需要钥匙才能加入他们，对吧？如何使用 2 个或更多密钥加入？我尝试复制这样的东西： select * from (select a. * from users a inner join deps b 关于 a.dep = b.dep 和 a.group = b.group ) a.id = b.id 上的内部连接 user_afy

我想将内部联接的数据保存在主题中，并将其用于外部联接。这是我的一个例子。

连接器属性：

....
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
mode=timestamp
query=select id, user, dep,tal, group,time from users
numeric.mapping=best_fit
table.types=TABLE
topic=users  
// I try use this with 1 or more fields but not worked  
transforms=createKey, extractInt  
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey  
transforms.createKey.fields=dep, group  
transforms.extractInt.type=org.apache.kafka.connect.transforms.ExtractField$Key  
transforms.extractInt.field=dep, group

standalone.properties

bootstrap.servers=localhost:9092  
key.converter=org.apache.kafka.connect.json.JsonConverter  
value.converter=org.apache.kafka.connect.json.JsonConverter  
key.converter.schemas.enable=false  
value.converter.schemas.enable=false  
offset.storage.file.filename=D:/tmp/connect.offsets  
plugin.path=D:/connector/lib

主题：

Topic users    
{"id":"0001", "user":"Alex", "dep":"ofi", "postal":170, group="ingen",time:"xxx"}    
{"id":"0002", "user":"Emy", "dep":"lab", "postal":170, group="itn",time:"xxx"}    
{"id":"0003", "user":"Lea", "dep":"lab", "postal":170, group="itn",time:"xxx"}    
{"id":"0004", "user":"Silva", "dep":"cent", "postal":170, group="ingen",time:"xxx"}    
{"id":"0005", "user":"Foxy", "dep":"cent", "postal":170, group="ete",time:"xxx"}    

topic user_afy
{"id":"0001", name="bask"}
{"id":"0001", name="Silf"}
{"id":"0002", name="BTT"}
{"id":"0005", name="butf"}


Topic deps  
{"id_dep":"1", "dep":"ofi", "sind"="worker", "group"="ingen."}  
{"id_dep":"2", "dep":"lab", "sind"="worker", "group"="iti."}  
{"id_dep":"3", "dep":"cent", "sind"="worker", "group"="etc."}

我的代码是官网的例子但是我无法测试

public static void main(String[] Args) {
        Properties props = new Properties();
        props.put(......);

    final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
        final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
        final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);
        final Consumed<String, JsonNode> consumed = Consumed.with(Serdes.String(), jsonSerde);
        StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, JsonNode> left = builder.stream("user", consumed);
        KTable<String, JsonNode> right = builder.table("deps", consumed);
        KStream<String, String> joined = left.join(right,
            (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue,
            Joined.with(Serdes.String(), jsonSerde, jsonSerde)
        );
//Edit
       joined.foreach((k, v) -> {
          System.out.println("key="+k+ ", val=" + v);
       });

}

输出，它会如何显示？要创建一个新主题，是否最好使用带有要保存为 json 格式的值的 hashmap？稍后我将创建自定义 Serdes

【问题讨论】：

你永远不会打印出蒸汽的输出。尝试使用 foreach 并查看输出。或者写个单元测试看看你做的是否正确
是的，我没有这样做，因为我无法创建记录，因为我无法创建密钥..
为什么不呢？您可以使用 map 函数从值中设置键。此外，最新的 kafka 2.4 让您无需两个流都具有相同的密钥即可加入

标签： java apache-kafka apache-kafka-streams apache-kafka-connect

【解决方案1】：

“我不能使用自己的密钥”是什么意思？在 Kafka Streams 中，您始终可以根据处理需要设置新密钥。

如果您想将数据读入 KTable，但您不能直接更改键。您需要将主题读取为 KStream，设置新密钥，并将 KStream 转换为 KTable（参见 Kafka Streams API: KStream to KTable）。

对于多个连续的连接，您可以将相应的操作“链接”在一起。

builder.stream("topic-1").selectKey(...).to("table-topic-1");
KTable t1 = builder.table("table-topic-1");

KStream firstJoinResult = builder.stream(...).selectKey(...).join(t1, ...).

builder.stream("topic-2").selectKey(...).to("table-topic-2");
KTable t2 = builder.table("table-topic-2");

firstJoinResult.selectKey(...).join(t2, ...).to("result-topic");

【讨论】：