实时计算Flink版：自定义表值函数（UDTF）

package com.hjc.test.blink.sql.udx;

import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.TableFunction;

public class SplitUdtf extends TableFunction<String> {

    // 可选，open方法可不编写。如果编写，则需要添加声明'import org.apache.flink.table.functions.FunctionContext;'。
    @Override
    public void open(FunctionContext context) {
        // ... ...
        }

    public void eval(String str) {
        String[] split = str.split("\\|");
        for (String s : split) {
            collect(s);
        }
    }

    // 可选，close方法可不编写。
    @Override
    public void close() {
        // ... ...
        }
}

• 返回值为Tuple
  实时计算Flink版支持使用Tuple1到Tuple25 ，定义1个字段到25个字段。用Tuple3来返回3个字段的UDTF示例如下。

  import org.apache.flink.api.java.tuple.Tuple3;
  import org.apache.flink.table.functions.TableFunction;
  
  // 使用Tuple作为返回值，一定要显式声明Tuple的泛型类型, 例如，String、Long和Integer。
  public class ParseUdtf extends TableFunction<Tuple3<String, Long, Integer>> {
  
  public void eval(String str) {
  String[] split = str.split(",");
  // 以下代码仅作示例，实际业务需要添加更多的校验逻辑。
  String first = split[0];
  long second = Long.parseLong(split[1]);
  int third = Integer.parseInt(split[2]);
  Tuple3<String, Long, Integer> tuple3 = Tuple3.of(first, second, third);
  collect(tuple3);
  }
  }

  说明 使用Tuple时，字段值不能为null，且最多只能存在25个字段。
• 返回值为Row
  使用Row来实现返回3个字段的UDTF示例如下。

  import org.apache.flink.table.types.DataType;
  import org.apache.flink.table.types.DataTypes;
  import org.apache.flink.table.functions.TableFunction;
  import org.apache.flink.types.Row;
  
  public class ParseUdtf extends TableFunction<Row> {
  
  public void eval(String str) {
  String[] split = str.split(",");
  String first = split[0];
  long second = Long.parseLong(split[1]);
  int third = Integer.parseInt(split[2]);
  Row row = new Row(3);
  row.setField(0, first);
  row.setField(1, second);
  row.setField(2, third);
  collect(row);
  }
  
  @Override
  // 如果返回值是Row，则必须重载实现getResultType方法，显式地声明返回的字段类型。
  public DataType getResultType(Object[] arguments, Class[] argTypes) {
  return DataTypes.createRowType(DataTypes.STRING, DataTypes.LONG, DataTypes.INT);
  }
  }

  说明 Row的字段值可以是null，但如果需要使用Row，必须重载实现getResultType方法。

CREATE FUNCTION parseUdtf AS 'com.alibaba.blink.sql.udtf.ParseUdtf';

• cross join
  左表的每一行数据都会关联上UDTF产出的每一行数据，如果UDTF不产出任何数据，则这1行不会输出。

  select S.id, S.content, T.a, T.b, T.c
  from input_stream as S,
  lateral table(parseUdtf(content)) as T(a, b, c);

• left join
  左表的每一行数据都会关联上UDTF产出的每一行数据，如果UDTF不产出任何数据，则这1行的UDTF的字段会用null值填充。

  说明 left join UDTF语句后面必须添加on true参数。

  select S.id, S.content, T.a, T.b, T.c
  from input_stream as S
  left join lateral table(parseUdtf(content)) as T(a, b, c) on true;