Kafka流联接表太大

Question

我想使用Kafka Streams API对带有KTable的KStream执行左联接，将表的一些字段添加到流中。

使用包含所有相关条目的较小版本的表（大约1300个条目），一切工作都很好。

使用整个表（大约200,000个条目）后，在获得KTable的Avro消息（GenericRecord）的相关字段的行中会得到一个NullPointerException。

当我在KSQL中执行相同的左联接操作时，从表中添加的字段为NULL。相关的联接键存在于表中，但在KSQL中查询它们需要大约。20秒后他们出现。

表是否太大，无法执行左联接？如果是这样的话，我可以做些什么来使连接使用整个表工作吗？

堆栈跟踪：

[myclass-68507371-7b8e-4bdc-8715-73d0307c9058-StreamThread-1] ERROR org.apache.kafka.streams.processor.internals.AssignedStreamsTasks - stream-thread [myclass-68507371-7b8e-4bdc-8715-73d0307c9058-StreamThread-1] Failed to process stream task 3_0 due to the following error:
java.lang.NullPointerException
  at MyClass.lambda$main$6(MyClass.java:184)
  at org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:73)
  at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
  at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
  at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
  at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
  at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
  at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
  at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
  at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
  at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)

该字段在Avro架构中是可为空的。

Answer 1

正如堆栈跟踪所指出的（错误源自MyClass.lambda$main$6(MyClass.java:184))，您自己的代码中的异常，必须是ValueJoiner实现。当您执行leftjoin()时，传入的“ktable”值可能是null。我假设您的代码没有正确处理null。如果您想要完全避免获得null并在没有相应的KTable记录时删除KStream记录，则可以使用join()而不是leftjoin()。