Не удается вернуть Struct из UDF KSQL - PullRequest
Новая
       54

Не удается вернуть Struct из UDF KSQL

0 голосов
/ 28 октября 2019

Я пытаюсь создать функцию UDF, которая принимает STRUCT в качестве параметра и возвращает структуру после внесения некоторых изменений во входную (обновление существующего поля + добавление новых полей), как показано в следующем коде: 

@UdfDescription(name = "myCustomUdf")
public class MyCustomUdf {

    @Udf(description = "do stuff")
    public Struct MyCustomUdf(@UdfParameter(schema = "struct <NAME VARCHAR, EMAIL VARCHAR>", value = "user") final Struct struct) {
        String processedEmail = struct.getString("EMAIL").toUpperCase();
        struct.put("EMAIL", processedEmail);
        return struct;
    }
}

Однако при развертывании моего пользовательского фляги я вижу в журналах KSQL следующее исключение, не могу найти в документации ничего, касающегося запрета этой функции, или я что-то здесь упускаю:

Этовыполнимо или нет, чтобы UDF вернул STRUCT? 

io.confluent.ksql.util.KsqlException: Could not load UDF method with signature: public org.apache.kafka.connect.data.Struct com.myudf.MyCustomUdf.MyCustomUdf(org.apache.kafka.connect.data.Struct)
    at io.confluent.ksql.function.UdfLoader.getReturnType(UdfLoader.java:373)
    at io.confluent.ksql.function.UdfLoader.addFunction(UdfLoader.java:280)
    at io.confluent.ksql.function.UdfLoader.lambda$handleUdfAnnotation$8(UdfLoader.java:217)
    at io.github.lukehutch.fastclasspathscanner.scanner.ScanSpec$9.lookForMatches(ScanSpec.java:1390)
    at io.github.lukehutch.fastclasspathscanner.scanner.ScanSpec.callMatchProcessors(ScanSpec.java:696)
    at io.github.lukehutch.fastclasspathscanner.FastClasspathScanner.scan(FastClasspathScanner.java:1606)
    at io.github.lukehutch.fastclasspathscanner.FastClasspathScanner.scan(FastClasspathScanner.java:1678)
    at io.github.lukehutch.fastclasspathscanner.FastClasspathScanner.scan(FastClasspathScanner.java:1704)
    at io.confluent.ksql.function.UdfLoader.loadUdfs(UdfLoader.java:145)
    at io.confluent.ksql.function.UdfLoader.lambda$load$2(UdfLoader.java:115)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
    at java.util.Iterator.forEachRemaining(Iterator.java:116)
    at java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
    at io.confluent.ksql.function.UdfLoader.load(UdfLoader.java:115)
    at io.confluent.ksql.rest.server.KsqlRestApplication.buildApplication(KsqlRestApplication.java:473)
    at io.confluent.ksql.rest.server.KsqlRestApplication.buildApplication(KsqlRestApplication.java:441)
    at io.confluent.ksql.rest.server.KsqlServerMain.createExecutable(KsqlServerMain.java:94)
    at io.confluent.ksql.rest.server.KsqlServerMain.main(KsqlServerMain.java:59)
Caused by: io.confluent.ksql.util.KsqlException: Type inference is not supported for: class org.apache.kafka.connect.data.Struct
    at io.confluent.ksql.util.SchemaUtil.handleParametrizedType(SchemaUtil.java:378)
    at io.confluent.ksql.util.SchemaUtil.lambda$getSchemaFromType$5(SchemaUtil.java:158)
    at io.confluent.ksql.util.SchemaUtil.getSchemaFromType(SchemaUtil.java:158)
    at io.confluent.ksql.util.SchemaUtil.getSchemaFromType(SchemaUtil.java:153)
    at io.confluent.ksql.function.UdfLoader.getReturnType(UdfLoader.java:366)
    ... 26 more

1 Ответ

0 голосов
/ 30 октября 2019

KSQL требует схему не только для входного аргумента, но и для выходного. В вашем случае обязательно укажите следующее в вашей аннотации @Udf: 

    @Udf(description = "do stuff", schema="STRUCT<name VARCHAR, email VARCHAR>")
    public Struct MyCustomUdf(@UdfParameter(schema = "struct <NAME VARCHAR, EMAIL VARCHAR>", value = "user") final Struct struct) {
       ...
    }

Похоже, что мы действительно упустили это из нашей документации! Я добавил проблему, чтобы убедиться, что она решена (https://github.com/confluentinc/ksql/issues/3699).

...