Kafka Stream和KGlobalTable连接问题

Question

我在加入一个KStream和一个GlobalKTable时遇到了一个问题，希望能得到您的帮助。

给定两个Kafka主题订单和客户:

订单

"1"     {"ID":"1","Name":"Myorder1","CustID":"100"}

"2"     {"ID":"2","Name":"MyOrder2","CustID":"200"}

客户

"100"   {"CustID":"100","CustName":"Customer1"}

"200"   {"CustID":"200","CustName":"Customer2"}

需求是用客户名称丰富订单流

"1"     {"ID":"1","Name":"Myorder1","CustID":"100","CustName":"Customer1"}

"2"     {"ID":"2","Name":"MyOrder2","CustID":"200","CustName":"Customer2"}}

我正在尝试以下操作：

null

java prettyprint-override">KStream<String, EnrichedOrder> enrichedstreams = orders.join(
    customers,
    new KeyValueMapper<String, Order, String>() {            
        @Override
        public String apply(String key, Order value) {
           return value.CustID;
        }
    },
    new ValueJoiner<Order,Customer, EnrichedOrder>() {
        @Override
        public EnrichedOrder apply(Order order, Customer customer) {
            EnrichedOrder eorder = new EnrichedOrder();
            eorder.CustID = order.CustID;
            eorder.CustName = customer.CustName;
            eorder.ID = order.ID;
            eorder.Name = order.Name;           
            return eorder;
        }
    }
);

Answer 1

让我们仔细看看你复制粘贴的内容：

在客户主题中：

"100"   {"CustID":"100","CustName":"Customer1"}

您可以注意到键是一个字符串，这个字符串包含双引号:“100”。通常，字符串键是不带双引号的。我宁愿看到：

 100    {"CustID":"100","CustName":"Customer1"}