User-defined Functions

User-defined functions are important features, because they significantly extend the expressiveness of Python Table API programs.

NOTE: Python UDF execution requires Python version (3.5, 3.6 or 3.7) with PyFlink installed. It’s required on both the client side and the cluster side.

Scalar Functions

It supports to use Python scalar functions in Python Table API programs. In order to define a Python scalar function, one can extend the base class ScalarFunction in pyflink.table.udf and implement an evaluation method. The behavior of a Python scalar function is defined by the evaluation method which is named eval. The evaluation method can support variable arguments, such as eval(*args).

The following example shows how to define your own Python hash code function, register it in the TableEnvironment, and call it in a query. Note that you can configure your scalar function via a constructor before it is registered:

  1. class HashCode(ScalarFunction):
  2.   def __init__(self):
  3.     self.factor = 12
  5.   def eval(self, s):
  6.     return hash(s) * self.factor
  8. table_env = BatchTableEnvironment.create(env)
  10. # configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
  11. table_env.get_config().get_configuration().set_string("taskmanager.memory.task.off-heap.size", '80m')
  13. # register the Python function
  14. table_env.register_function("hash_code", udf(HashCode(), DataTypes.BIGINT(), DataTypes.BIGINT()))
  16. # use the Python function in Python Table API
  17. my_table.select("string, bigint, bigint.hash_code(), hash_code(bigint)")
  19. # use the Python function in SQL API
  20. table_env.sql_query("SELECT string, bigint, hash_code(bigint) FROM MyTable")

Note If not using RocksDB as state backend, you can also configure the python worker to use the managed memory of taskmanager by setting python.fn-execution.memory.managed to be true. Then there is no need to set the the configuration taskmanager.memory.task.off-heap.size.

It also supports to use Java/Scala scalar functions in Python Table API programs.

  1. '''
  2. Java code:
  4. // The Java class must have a public no-argument constructor and can be founded in current Java classloader.
  5. public class HashCode extends ScalarFunction {
  6.   private int factor = 12;
  8.   public int eval(String s) {
  9.       return s.hashCode() * factor;
  10.   }
  11. }
  12. '''
  14. table_env = BatchTableEnvironment.create(env)
  16. # configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
  17. table_env.get_config().get_configuration().set_string("taskmanager.memory.task.off-heap.size", '80m')
  19. # register the Java function
  20. table_env.register_java_function("hash_code", "my.java.function.HashCode")
  22. # use the Java function in Python Table API
  23. my_table.select("string.hash_code(), hash_code(string)")
  25. # use the Java function in SQL API
  26. table_env.sql_query("SELECT string, bigint, hash_code(string) FROM MyTable")

Note If not using RocksDB as state backend, you can also configure the python worker to use the managed memory of taskmanager by setting python.fn-execution.memory.managed to be true. Then there is no need to set the the configuration taskmanager.memory.task.off-heap.size.

There are many ways to define a Python scalar function besides extending the base class ScalarFunction. The following examples show the different ways to define a Python scalar function which takes two columns of bigint as the input parameters and returns the sum of them as the result.

  1. # option 1: extending the base class `ScalarFunction`
  2. class Add(ScalarFunction):
  3.   def eval(self, i, j):
  4.     return i + j
  6. add = udf(Add(), [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  8. # option 2: Python function
  9. @udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT())
  10. def add(i, j):
  11.   return i + j
  13. # option 3: lambda function
  14. add = udf(lambda i, j: i + j, [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  16. # option 4: callable function
  17. class CallableAdd(object):
  18.   def __call__(self, i, j):
  19.     return i + j
  21. add = udf(CallableAdd(), [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT())
  23. # option 5: partial function
  24. def partial_add(i, j, k):
  25.   return i + j + k
  27. add = udf(functools.partial(partial_add, k=1), [DataTypes.BIGINT(), DataTypes.BIGINT()],
  28.           DataTypes.BIGINT())
  30. # register the Python function
  31. table_env.register_function("add", add)
  32. # use the function in Python Table API
  33. my_table.select("add(a, b)")

Table Functions

Similar to a Python user-defined scalar function, a user-defined table function takes zero, one, or multiple scalar values as input parameters. However in contrast to a scalar function, it can return an arbitrary number of rows as output instead of a single value. The return type of a Python UDTF could be of types Iterable, Iterator or generator.

The following example shows how to define your own Python multi emit function, register it in the TableEnvironment, and call it in a query.

  1. class Split(TableFunction):
  2.     def eval(self, string):
  3.         for s in string.split(" "):
  4.             yield s, len(s)
  6. env = StreamExecutionEnvironment.get_execution_environment()
  7. table_env = StreamTableEnvironment.create(env)
  8. my_table = ...  # type: Table, table schema: [a: String]
  10. # configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
  11. table_env.get_config().get_configuration().set_string("taskmanager.memory.task.off-heap.size", '80m')
  13. # register the Python Table Function
  14. table_env.register_function("split", udtf(Split(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.INT()]))
  16. # use the Python Table Function in Python Table API
  17. my_table.join_lateral("split(a) as (word, length)")
  18. my_table.left_outer_join_lateral("split(a) as (word, length)")
  20. # use the Python Table function in SQL API
  21. table_env.sql_query("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a)) as T(word, length)")
  22. table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE")

Note If not using RocksDB as state backend, you can also configure the python worker to use the managed memory of taskmanager by setting python.fn-execution.memory.managed to be true. Then there is no need to set the the configuration taskmanager.memory.task.off-heap.size.

It also supports to use Java/Scala table functions in Python Table API programs.

  1. '''
  2. Java code:
  4. // The generic type "Tuple2<String, Integer>" determines the schema of the returned table as (String, Integer).
  5. // The java class must have a public no-argument constructor and can be founded in current java classloader.
  6. public class Split extends TableFunction<Tuple2<String, Integer>> {
  7.     private String separator = " ";
  9.     public void eval(String str) {
  10.         for (String s : str.split(separator)) {
  11.             // use collect(...) to emit a row
  12.             collect(new Tuple2<String, Integer>(s, s.length()));
  13.         }
  14.     }
  15. }
  16. '''
  18. env = StreamExecutionEnvironment.get_execution_environment()
  19. table_env = StreamTableEnvironment.create(env)
  20. my_table = ...  # type: Table, table schema: [a: String]
  22. # configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
  23. table_env.get_config().get_configuration().set_string("taskmanager.memory.task.off-heap.size", '80m')
  25. # Register the java function.
  26. table_env.register_java_function("split", "my.java.function.Split")
  28. # Use the table function in the Python Table API. "as" specifies the field names of the table.
  29. my_table.join_lateral("split(a) as (word, length)").select("a, word, length")
  30. my_table.left_outer_join_lateral("split(a) as (word, length)").select("a, word, length")
  32. # Register the python function.
  34. # Use the table function in SQL with LATERAL and TABLE keywords.
  35. # CROSS JOIN a table function (equivalent to "join" in Table API).
  36. table_env.sql_query("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a)) as T(word, length)")
  37. # LEFT JOIN a table function (equivalent to "left_outer_join" in Table API).
  38. table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE")

Note If not using RocksDB as state backend, you can also configure the python worker to use the managed memory of taskmanager by setting python.fn-execution.memory.managed to be true. Then there is no need to set the the configuration taskmanager.memory.task.off-heap.size.

Like Python scalar functions, you can use the above five ways to define Python TableFunctions.

Note The only difference is that the return type of Python Table Functions needs to be an iterable, iterator or generator.

  1. # option 1: generator function
  2. @udtf(input_types=DataTypes.BIGINT(), result_types=DataTypes.BIGINT())
  3. def generator_func(x):
  4.       yield 1
  5.       yield 2
  7. # option 2: return iterator
  8. @udtf(input_types=DataTypes.BIGINT(), result_types=DataTypes.BIGINT())
  9. def iterator_func(x):
  10.       return range(5)
  12. # option 3: return iterable
  13. @udtf(input_types=DataTypes.BIGINT(), result_types=DataTypes.BIGINT())
  14. def iterable_func(x):
  15.       result = [1, 2, 3]
  16.       return result
  18. table_env.register_function("iterable_func", iterable_func)
  19. table_env.register_function("iterator_func", iterator_func)
  20. table_env.register_function("generator_func", generator_func)