我正在尝试创建一个流为select(CSAS),该流已成功创建,但是当我尝试推送消息时,出现以下异常。
Caused by: org.apache.kafka.connect.errors.DataException: Struct schemas do not match.
at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:247)
at org.apache.kafka.connect.data.Struct.put(Struct.java:216)
at io.confluent.ksql.serde.GenericRowSerDe$GenericRowSerializer.serialize(GenericRowSerDe.java:116)
at io.confluent.ksql.serde.GenericRowSerDe$GenericRowSerializer.serialize(GenericRowSerDe.java:93)
at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:162)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:102)
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
下面是来自ksql-cli的主流,持久性流和udf函数的详细信息,不确定为什么架构不兼容,因为您可以在processed
流下面看到一个带有article
字段的模式与UDF函数返回的值完全相同,我在这里丢失了吗?
ksql> create stream main_stream ( article struct< _id VARCHAR, title VARCHAR, text VARCHAR, action VARCHAR, url VARCHAR, feed_id VARCHAR, mode VARCHAR, score INTEGER, published_at VARCHAR, retrieved_at VARCHAR> ) with (KAFKA_TOPIC='articles', value_format='JSON');
ksql> create stream processed as select test(article) article from main_stream;
ksql> describe processed;
Name : processed
Field | Type
-------------------------------------------------------------------------------------------------------------------------------------------------------------
ROWTIME | BIGINT (system)
ROWKEY | VARCHAR(STRING) (system)
ARTICLE | STRUCT<_ID VARCHAR(STRING), RAW_TITLE VARCHAR(STRING), RAW_TEXT VARCHAR(STRING), PROCESSED_TITLE VARCHAR(STRING), PROCESSED_TEXT VARCHAR(STRING)>
-------------------------------------------------------------------------------------------------------------------------------------------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;
ksql> show queries;
Query ID | Kafka Topic | Query String
--------------------------------------------------------------------------------------------------------------------------------------------------------------
CSAS_processed_20 | processed | CREATE STREAM processed WITH (REPLICAS = 1, PARTITIONS = 1, KAFKA_TOPIC = 'processed') AS SELECT TEST(MAIN_STREAM.ARTICLE) "ARTICLE"
FROM MAIN_STREAM MAIN_STREAM;
--------------------------------------------------------------------------------------------------------------------------------------------------------------
ksql> describe function test;
Name : TEST
Overview : test udf
Type : scalar
Jar : /Users/ktawfik/libs/custom-udf.jar
Variations :
Variation : TEST(article STRUCT<_ID VARCHAR, TITLE VARCHAR, TEXT VARCHAR, ACTION VARCHAR, URL VARCHAR, FEED_ID VARCHAR, MODE VARCHAR, SCORE INT, PUBLISHED_AT VARCHAR, RETRIEVED_AT VARCHAR>)
Returns : STRUCT<_ID VARCHAR, RAW_TITLE VARCHAR, RAW_TEXT VARCHAR, PROCESSED_TITLE VARCHAR, PROCESSED_TEXT VARCHAR>
Description : test
article : A complete article object
也低于我使用的UDF代码
@Udf(description = "test",
schema = "struct< _id VARCHAR, raw_title VARCHAR, raw_text VARCHAR, processed_title VARCHAR, processed_text VARCHAR>")
public Struct processDocument(
@UdfParameter(
schema = "struct< _id VARCHAR, title VARCHAR, text VARCHAR, action VARCHAR, url VARCHAR, feed_id VARCHAR, mode VARCHAR, score INTEGER, published_at VARCHAR, retrieved_at VARCHAR>",
value = "article",
description = "A complete article object") Struct struct) {
Schema ARTICLE_SCHEMA = SchemaBuilder.struct()
.field("_id", Schema.STRING_SCHEMA)
.field("raw_title", Schema.STRING_SCHEMA)
.field("raw_text", Schema.STRING_SCHEMA)
.field("processed_title", Schema.STRING_SCHEMA)
.field("processed_text", Schema.STRING_SCHEMA)
.build();
Struct proStruct = new Struct(ARTICLE_SCHEMA);
proStruct.put("_id", "1234");
proStruct.put("raw_title", "RAW_TITLE___1234");
proStruct.put("raw_text", "RAW_TEXT___1234");
proStruct.put("processed_title", "TITLE____1234");
proStruct.put("processed_text", "TEXT____1234");
System.out.println(proStruct);
// Struct{_id=1234,raw_title=RAW_TITLE___1234,raw_text=RAW_TEXT___1234,processed_title=TITLE____1234,processed_text=TEXT____1234}
return proStruct;
}
我能够找出问题并解决,基本上是KSQL引擎将架构字段转换为大写的事实,因此,当我发送小写的字段时,它无法匹配它,这尚不清楚在文档中。
解决方法是我必须拥有:
@UDF
注释中的架构字段。@UDF
注释中的架构字段中的所有字段(名称和类型)完全匹配。代码终于看起来像:
@Udf(description = "test",
schema = "struct< _ID VARCHAR, RAW_TITLE VARCHAR, RAW_TEXT VARCHAR, PROCESSED_TITLE VARCHAR, PROCESSED_TEXT VARCHAR>")
public Struct processDocument(
@UdfParameter(
schema = "struct< _id VARCHAR, title VARCHAR, text VARCHAR, action VARCHAR, url VARCHAR, feed_id VARCHAR, mode VARCHAR, score INTEGER, published_at VARCHAR, retrieved_at VARCHAR>",
value = "article",
description = "A complete article object") Struct struct) {
Schema ARTICLE_SCHEMA = SchemaBuilder.struct()
.field("_ID", Schema.STRING_SCHEMA)
.field("RAW_TITLE", Schema.STRING_SCHEMA)
.field("RAW_TEXT", Schema.STRING_SCHEMA)
.field("PROCESSED_TITLE", Schema.STRING_SCHEMA)
.field("PROCESSED_TEXT", Schema.STRING_SCHEMA)
.build();
Struct proStruct = new Struct(ARTICLE_SCHEMA);
proStruct.put("_ID", "1234");
proStruct.put("RAW_TITLE", "RAW_TITLE___1234");
proStruct.put("RAW_TEXT", "RAW_TEXT___1234");
proStruct.put("PROCESSED_TITLE", "TITLE____1234");
proStruct.put("PROCESSED_TEXT", "TEXT____1234");
System.out.println(proStruct);
return proStruct;
}