diff --git a/docs/dev/table/python/python_udfs.md b/docs/dev/table/python/python_udfs.md index ea01cd0ca3faba..58bf528d3452d7 100644 --- a/docs/dev/table/python/python_udfs.md +++ b/docs/dev/table/python/python_udfs.md @@ -48,6 +48,9 @@ class HashCode(ScalarFunction): table_env = BatchTableEnvironment.create(env) +# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory. +table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m') + # register the Python function table_env.register_function("hash_code", udf(HashCode(), DataTypes.BIGINT(), DataTypes.BIGINT())) @@ -58,6 +61,8 @@ my_table.select("string, bigint, bigint.hash_code(), hash_code(bigint)") table_env.sql_query("SELECT string, bigint, hash_code(bigint) FROM MyTable") {% endhighlight %} +Note If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**. + It also supports to use Java/Scala scalar functions in Python Table API programs. {% highlight python %} @@ -76,6 +81,9 @@ public class HashCode extends ScalarFunction { table_env = BatchTableEnvironment.create(env) +# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory. +table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m') + # register the Java function table_env.register_java_function("hash_code", "my.java.function.HashCode") @@ -86,6 +94,8 @@ my_table.select("string.hash_code(), hash_code(string)") table_env.sql_query("SELECT string, bigint, hash_code(string) FROM MyTable") {% endhighlight %} +Note If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**. + 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. @@ -145,6 +155,9 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] +# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory. +table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m') + # register the Python Table Function table_env.register_function("split", udtf(Split(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.INT()])) @@ -158,6 +171,8 @@ table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE {% endhighlight %} +Note If not using RocksDB as state backend, you can configure the python worker to use the managed memory of +taskmanager by setting **python.fn-execution.memory.managed** to be **true**. It also supports to use Java/Scala table functions in Python Table API programs. {% highlight python %} @@ -182,6 +197,9 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] +# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory. +table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m') + # Register the java function. table_env.register_java_function("split", "my.java.function.Split") @@ -198,6 +216,8 @@ table_env.sql_query("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a) table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE") {% endhighlight %} +Note If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**. + 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. diff --git a/docs/dev/table/python/vectorized_python_udfs.md b/docs/dev/table/python/vectorized_python_udfs.md index 8eec40bf497bc4..28337f17f4ef94 100644 --- a/docs/dev/table/python/vectorized_python_udfs.md +++ b/docs/dev/table/python/vectorized_python_udfs.md @@ -54,6 +54,9 @@ def add(i, j): table_env = BatchTableEnvironment.create(env) +# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory. +table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m') + # register the vectorized Python scalar function table_env.register_function("add", add) @@ -63,3 +66,5 @@ my_table.select("add(bigint, bigint)") # use the vectorized Python scalar function in SQL API table_env.sql_query("SELECT add(bigint, bigint) FROM MyTable") {% endhighlight %} + +Note If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**.