ksql:udf不接受参数(string,string)

pu82cl6c  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(447)

我在尝试使用自定义项设置etl管道时遇到了ksql的一个问题。在etl过程中的某个时刻,我需要从数据中的描述字段(varchar)中隔离特定的信息。上下文的虚构示例:
description=“物种=狗。性别=男性。颜色=金发。年龄=10。”(实际数据的格式相同)
我已经编写了一个简单的自定义项来隔离任何需要的信息。看起来是这样的:

package com.my.package;

/**IMPORTS**/
import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;

/**ClASS DEFINITION**/
@UdfDescription(name = "extract_from_description",
                author = "Me",
                version = "0.0.1",
                description = "Given a description and a request for information, isolates and returns the requested information. Pass requested tag as 'tag='".) 
public class Extract_From_Description {

    @Udf(description = "Given a description and a request for information, isolates and returns the requested information. Pass requested tag as 'tag='.)
    public String extract_from_description(final String description, final String request) {
        return description.split(request)[1].split("\\.")[0];
    }
}

我可以很好的上传和注册这个函数,当我运行的时候它被正确的列出和描述了:

ksql> list functions;
ksql> describe function EXTRACT_FROM_DESCRIPTION;

我这样调用函数来创建一个新流:

CREATE STREAM result AS
    SELECT recordId,
           OtherVariables,
           EXTRACT_FROM_DESCRIPTION(description, 'species=') AS species
    FROM parent_stream
    EMIT CHANGES;

有个错误我搞不懂:
函数“extract \u from \u description”不接受参数(string,string)。有效的替代方案包括:
显然ksql不能正确解释函数的输入应该是什么(看起来它不需要输入?),我也不知道为什么。我已经通读了文档,看看我是否以一种奇怪的方式定义了我的函数,但是没有发现示例和我的函数之间的任何差异。我注意到应该有几种方法来定义函数的输入,并尝试了所有方法,但结果总是一样的。
我使用maven为这个函数创建jar文件(jdk1.8.0\u201)。有人能帮我弄清楚发生了什么事吗?
热释光;dr:my ksql udf不接受类型(string,string)的输入,即使函数指定输入应为类型(string,string)

2ul0zpep

2ul0zpep1#

找到了问题,在这里回答任何可能遇到同样问题的人。您需要使用@udfparameter指定参数,如下所示:

import io.confluent.ksql.function.udf.UdfParameter; // add this to the list of imports

// add @UdfParameter(name) to each input variable
public String extract_from_description(@UdfParameter(value = "description") final String description, @UdfParameter(value = "request") final String request){

  function body

}

相关问题