Flink SQL : UDTF传递行类型参数

Question

CREATE TABLE user_log (
    data ROW(id String,user_id String,class_id String)
) WITH (
    'connector.type' = 'kafka',
    ...
);

INSERT INTO sink
SELECT * FROM user_log as tab,
LATERAL TABLE(splitUdtf(tab.data)) AS T(a,b,c);

UDTF代码：

public void eval(Row data) {...}

eval方法只能传递行类型参数吗？我想获得SQL中的行键，如id、user_id、class_id，但是java中的行键是索引(如0、1、2).How，我做了吗？谢谢!

Answer 1

您的sql能够直接将kafka数据转换为表行吗？也许不会。行是DataStream级别的类型，而不是TableAPI&SQL中的类型。

如果从kafka收到的数据是json格式的，则可以在fllink sql中使用DDL语句，或者使用Connector API直接提取json中的字段，只要json是键值格式的。