UDTF

A UDTF needs to implement the eval method in the TableFunction class. The open and close methods are optional. The following Java code is an example:

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> {

    // The open method is optional. To use the open method, you must add 'import org.apache.flink.table.functions.FunctionContext;' to the code. 
    @Override
    public void open(FunctionContext context) {
        // ... ...
        }

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

    // The close method is optional. 
    @Override
    public void close() {
        // ... ...
        }
}

A UDTF can also convert the output result from a single column to multiple columns. If you want a UDTF to return multiple columns, declare the return value as a tuple or row. The following examples show how to declare a return value as a tuple and how to declare a return value as a row.

• Declare the return value as a tuple.
  Realtime Compute for Apache Flink supports Tuple1 to Tuple25, which define 1 to 25 fields. The following example is a UDTF that uses Tuple3 to return three fields:

  import org.apache.flink.api.java.tuple.Tuple3;
  import org.apache.flink.table.functions.TableFunction;
  
  // If the return value is declared as a tuple, you must explicitly declare the generic types of the tuple, such as STRING, LONG, and INTEGER in this example. 
  public class ParseUdtf extends TableFunction<Tuple3<String, Long, Integer>> {
  
  public void eval(String str) {
  String[] split = str.split(",");
  // The following code is used for reference only. In actual scenarios, you must add code that implements verification logic. 
  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);
  }
  }

  Note If the return value is declared as a tuple, column values cannot be null and a maximum of 25 columns are allowed.
• Declare the return value as a row.
  For example, you can enable this feature to return three columns. The sample code is an example.

  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
  // If the return value is declared as a row, you must overload the getResultType method to explicitly declare the data type of the return value. 
  public DataType getResultType(Object[] arguments, Class[] argTypes) {
  return DataTypes.createRowType(DataTypes.STRING, DataTypes.LONG, DataTypes.INT);
  }
  }

  Note If the return value is declared as a row, the field value can be null. However, you must overload the getResultType method.

A UDTF supports CROSS JOIN and LEFT JOIN. When you use a UDTF, you must add the keywords LATERAL and TABLE. You must also specify an alias for the UDTF, such as ParseUdtf in the preceding code.

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

• cross join
  Each row in the left table is joined with a row of data that is generated by the UDTF. If the UDTF does not generate any data for a row, the row is not returned.

  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
  Each row in the left table is joined with a row of data that is generated by the UDTF. If the UDTF does not generate any data for a row, the UDTF fields in the row are filled with null.

  Note A LEFT JOIN statement that uses a UDTF must end with 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;