我有两个KTable对象:

KTable<Long, byte[]> firstTable = builder.table("firstTopic", Consumed.with(Serdes.Long(), Serdes.ByteArray()));

 KTable<Long, byte[]> secondTable = builder.table("secondTopic",
        Consumed.with(Serdes.Long(), Serdes.ByteArray()));


之后,我想加入这两个表:

firstTable.leftJoin(secondTable,
            (leftValue, rightValue) -> {
            try {
                return utils.serializeNetwork(utils.deserializeNetwork(leftValue));
            } catch (IOException e) {
                e.printStackTrace();
                return null;
            }
            }
          )


因此,我有两个表,并将它们连接到一个表中,并且我希望通过每个键将结果表存储在kafka状态存储中,但是我不知道该怎么做。

最佳答案

您可以通过在Materialized上指定leftJoin参数并为状态存储指定名称来将实例化强制到本地存储中。

firstTable.leftJoin(..., Materialized.as("my-store-name"));

关于java - 如何联接两个KTable并将结果ktable写入状态存储,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/53616769/

10-10 18:03