apache-kafka - Kafka流处理过程中处理数据的异常
问题描述
我正在使用以下代码处理 Kafka 流。我从 JSON obj 检查条件 if 的过滤条件"UserID":"1"
。请参考下面的代码
builder.<String,String>stream(Serdes.String(), Serdes.String(), topic)
.filter(new Predicate <String, String>() {
String userIDCheck = null;
@Override
public boolean test(String key, String value) {
try {
JSONObject jsonObj = new JSONObject(value);
userIDCheck = jsonObj.get("UserID").toString();
System.out.println("userIDCheck: " + userIDCheck);
} catch (JSONException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return userIDCheck.equals("1");
}
})
.to(streamouttopic);
值:{"UserID":"1","Address":"XXX","AccountNo":"989","UserName":"Stella","AccountType":"YYY"}
我收到以下错误:
Exception in thread "SampleStreamProducer-4eecc3ab-858c-44a4-9b8c-5ece2b4ab21a-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=testtopic1, partition=0, offset=270
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:203)
at org.apache.kafka.streams.processor.internals.StreamThread.processAndPunctuate(StreamThread.java:679)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:557)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.ByteArraySerializer / value: org.apache.kafka.common.serialization.ByteArraySerializer) is not compatible to the actual key or value type (key type: unknown because key is null / value type: java.lang.String). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:91)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:82)
at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:47)
at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:82)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:189)
... 3 more
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to [B
at org.apache.kafka.common.serialization.ByteArraySerializer.serialize(ByteArraySerializer.java:21)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:89)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:76)
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:87)
上面的流代码中的值和条件都很好,我不明白为什么在执行流代码时它会给出这个异常。
解决方案
报告的问题应仅适用于 Kafka 2.0 及更早版本。从 2.1.0 版本开始,Kafka Streams 支持“serde push down”,to()
操作者应该从上游继承正确的 serdes(参见https://issues.apache.org/jira/browse/KAFKA-7456)。
对于 Kafka 2.0 及更早版本,您必须to()
明确地为操作指定正确的 Serdes。否则,它使用默认的Serdes(因为语义或 serde overwriteStreamsConfig
是ByteArraySerde
每个运算符“插入式覆盖”)——并且String
不能强制转换为byte[]
.
你需要做:
.to(streamoutputtopic, Produced.with(Serdes.String(), Serdes.String()));
对于Produced
尚未使用参数的旧版本(1.0 之前),代码将是:
.to(Serdes.String(), Serdes.String(), streamoutputtopic);
推荐阅读
- javascript - 有没有办法检查用户是否正在流式传输并显示它?
- python - 在python中将原始二进制数据转换为无符号整数
- python - 如何显示产品的 Django 其他嵌套评论?
- json - cts:near-query Marklogic on json 文档
- android - 为什么异步任务发送不明确的文件名?
- vue.js - 当我收到 vue_1.ref 不是函数错误时,如何使用 uve-recaptcha-v3 将 Recaptcha v3 与 VueJS 一起使用?
- python - 在python中解析错误时出现意外的EOF
- android - 从任何保留 backstack 的地方转到特定片段
- python - 有没有办法在执行时取消 Airflow dag 运行?
- python - setup.py:“描述”中的换行符