使用我想要拆分并发送到两个不同主题的 kafkastreams 使用 json

Question

我有一个来自 kafka 主题的大型 json，我正在将其转换为 Java 对象，以便在查找数据库中提取我需要的值。有些记录中会有一系列缺陷，我需要捕获这些缺陷并将其发送到不同的主题，以便它们最终可以出现在数据库中自己的表中。使用接收器连接器将值插入到数据库中，这就是我们使用多个主题的原因。

我发现了分支和拆分，但看起来这更多是为了确定所使用的记录应该转到哪个主题，而不是将记录的不同部分发送到不同的主题。有没有办法做到这一点，或者我是否需要在某个地方改变我的架构。

    @Autowired
    void buildPipeline(StreamsBuilder builder) throws Exception{
        KStream<String, String> messageStream = builder.stream(inputTopic, Consumed.with(STRING_SERDE, STRING_SERDE));
                logger.info("started consumer");
        System.out.println(messageStream.toString());

        KStream<String, String> auditRecords = messageStream
                  .map((key, value) -> {
                      try {
                          return new KeyValue<>("", mapStringToAuditOutput(value));
                      } catch (JsonProcessingException e) {
                          e.printStackTrace();
                      }
                      return null;
                  });
     auditRecords.to(outputTopic);

    }
    public String mapStringToAuditOutput(String input) throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        AuditOutput auditResults = null;
        try {
            auditResults=  mapper.readValue(input, AuditOutput.class);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        //check for multiple defects
        if(auditResults.getCaseObject().getDefects() != null){
            //send defects to separate topic

        }
        String auditJson = JsonFlattener.flatten(mapper.writeValueAsString(auditResults));
        System.out.println(auditJson);
        return auditJson;
    }

Answer 1

找到了分支和分割，但看起来这更多的是为了确定消费的记录应该转到哪个主题

正确。您需要在分支之前过滤+map/mapValues，以将部分/整个事件发送到不同的主题

更具体地说，创建中间KStream实例并多次使用to()

例如，

// Change the value serde to use your JSON class
KStream<String, AuditOutput> auditRecords = messageStream 
  .mapValues(value -> {
      try {
          // input Stream can be <String, String> but this step would be automatic if using a JSONSerde
          return mapStringToAuditOutput(value));
      } catch (JsonProcessingException e) {
          e.printStackTrace();
      }
      return null;
  }).filter((key, value) -> Objects.nonNull(value)); // remove bad JSON events
Map<String, KStream<String, AuditOutput>> branches = auditRecords.split(Named.as("Branch-"))
    .branch((key, value) -> value.getCaseObject().getDefects() != null,  /* first predicate  */
         Branched.as("Defects"))
    .defaultBranch(Branched.as("NoDefects"))              /* default branch */
);
branches.get("Branch-Defects").to(...)
branches.get("Branch-NoDefects").mapValues(... flatten ... ).to(...)