如何将记录拆分为不同的流，从一个主题到不同的流？

Question

我有一个包含不同大小记录的单一源CSV文件，它将每个记录推送到一个源主题中。我想将这些记录从源主题拆分成不同的KStreams/KTables。我有一个用于一个表加载的管道，在这里我将记录从源主题以分隔格式推送到stream1中，然后将记录推送到另一个AVRO格式的流中，再将该流推送到JDBC接收器连接器中，该连接器将记录推送到MySQL数据库中。管道需要相同。但我希望将不同表的记录推送到一个源主题中,然后根据一个值将这些记录拆分为不同的流。这可能吗？我试着想办法做到这一点，但做不到。我是否可以改进管道，或者使用KTable而不是KStreams或任何其他修改？

我有一个不同的MySQL表test2具有不同的模式，该表的记录也存在于source.csv文件中。由于模式不同，我无法遵循test1的当前管道将数据插入test2表。

示例-在CSV源文件中,

线路1-9,atm,mun,ronaldo线路2-10,atm,mun,bravo,num2线路3-11,atm,sign,bravo,Sick

如果col4==C罗，如果col4==bravo和col3==mun，则转到表1，如果col4==bravo和col3==sign转到表3

我对Kafka很陌生，从前一周开始开发Kafka。

Answer 1

您可以编写一个分离的Kafka Streams应用程序，使用KStream#branch()运算符将记录从输入主题拆分到不同的KStream或输出主题：

KStream<K, V>[] branches = streamsBuilder.branch(
        (key, value) -> {filter logic for topic 1 here},
        (key, value) -> {filter logic for topic 2 here},
        (key, value) -> true//get all messages for this branch
);

// KStream branches[0] records for logic 1
// KStream branches[1] records for logic 2
// KStream branches[2] records for logic 3

或者可以像这样手动分支KStream：

KStream<K, V> inputKStream = streamsBuilder.stream("your_input_topic", Consumed.with(keySerde, valueSerdes));

inputKStream
        .filter((key, value) -> {filter logic for topic 1 here})
        .to("your_1st_output_topic");

inputKStream
        .filter((key, value) -> {filter logic for topic 2 here})
        .to("your_2nd_output_topic");
...