一对多 KStream-KStream 加入答案

【问题标题】：One to Many KStream-KStream join一对多 KStream-KStream 加入
【发布时间】：2017-10-25 14:22:35
【问题描述】：

如何在两个 kafka KStream 之间执行一对多连接？下面给出的代码以一对一的方式连接了两个 Kafka KStream。有人可以指导如何在 KStream 之间执行一对多连接吗？主题中接收到的数据是泛型在主题中写入的数据具有以下形式 {"来自订单":"test1:,"来自订单项":"test2"} {"来自订单":"test1:,"来自订单项":"test3"}

是否可以获取这种格式的数据： {"从订单":"test1,{"从订单项":"test2"},{"从订单项":"test3"}}

public class ConsumerThreadPool {

private static final String TOPIC = "jre1";
private static final String NEXTTOPIC ="Kafka";
private static final String FINALTOPIC="jvm1";
private static final Integer NUM_THREADS = 1;
final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();

final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);
final Serde<String> stringSerde = Serdes.String();


int threadNumber = 0;
@Autowired
private ConsumerConfigFactory consumerConfigFactory;

@SuppressWarnings("unused")
private ConsumerConnector consumer;
private ExecutorService threadPool;

public ConsumerThreadPool() {
    threadPool = Executors.newFixedThreadPool(NUM_THREADS);
}

@PostConstruct
public void startConsuming() {
    ConsumerConfig consumerConfig = consumerConfigFactory.getConsumerConfig();
    consumer = createJavaConsumerConnector(consumerConfig);
    KStreamBuilder builder = new KStreamBuilder();
    /* KTable<String,JsonNode> message = builder.table(stringSerde,jsonSerde,TOPIC);


    KTable<String,JsonNode> orderstream = message

            .filter((k,v)-> v.path("table").asText().equals("TEST.S_ORDER")
                    );              
    KTable<String,JsonNode> orderlist=message.filter((k,v)-> v.path("table").asText().equals("TEST.S_ORDER_ITEM"));
    orderstream.to(stringSerde,jsonSerde,FINALTOPIC);      
    orderlist.to(stringSerde,jsonSerde,FINALTOPIC);    */ 
    KStream<String,JsonNode>streams=builder.stream(TOPIC);

    KStream<String,JsonNode> orderstream=streams.filter((k,v)-> v.path("table").asText().equals("TEST.S_ORDER"))
            .map((k,v)->KeyValue.pair(v.path("after").path("ROW_ID").asText(),v));




    KStream<String, JsonNode> orderlist=streams.filter((k,v)-> v.path("table").asText().equals("TEST.S_ORDER_ITEM"))
            .map((k,v)->KeyValue.pair(v.path("after").path("ORDER_ID").asText(),v));





    KStream<String,JsonNode> nextstream =orderstream.join(orderlist,(new ValueJoiner<JsonNode,JsonNode,JsonNode>(){
        @Override
        public JsonNode apply(JsonNode first,JsonNode second){
            ObjectNode jNode = JsonNodeFactory.instance.objectNode();
            return jNode.put("from order",first.get("op_type").textValue())
                    .put("from orderitem",second.get("op_type").textValue() );
        }
    }),JoinWindows.of(TimeUnit.SECONDS.toMillis(30)),stringSerde,jsonSerde,jsonSerde);

    nextstream.to(stringSerde,jsonSerde,FINALTOPIC);  
    KafkaStreams stream=new KafkaStreams(builder, consumerConfigFactory.getConsumeConfig());
    stream.start();
    consume();
    stream.close();
}

public void consume() {



    @SuppressWarnings("resource")
    KafkaConsumer<String,String> consumer = new KafkaConsumer<>(consumerConfigFactory.createConsume());
    consumer.subscribe(Arrays.asList(FINALTOPIC));

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        if(!records.isEmpty()){
            System.out.println("ConsumerRecords object created: "+records);
            threadPool.submit(new MessageConsumer(records, threadNumber));
            threadNumber++;
        }

    }

}

}

【问题讨论】：

标签： java kafka-consumer-api apache-kafka-streams

【解决方案1】：

正如您已经提到的，KStream-KStream 已经是一对多连接。您似乎想将唯一键的所有连接结果聚合到一条记录中。

您可以申请.groupByKey().aggregate() 来执行此操作。聚合函数使用空 JSON 进行初始化，每次新的连接结果到达时都会向 JSON 添加新记录。

【讨论】：