From cbd3259908d23c419e97595704e59af2ce970cab Mon Sep 17 00:00:00 2001 From: "acqua.csq" Date: Wed, 8 Jul 2020 20:28:49 +0800 Subject: [PATCH 1/2] [FLINK-18526][python][docs] Add configuration of Python UDF to use Managed Memory in the doc of Pyflink. --- docs/dev/table/python/python_udfs.md | 12 ++++++++++++ docs/dev/table/python/vectorized_python_udfs.md | 3 +++ 2 files changed, 15 insertions(+) diff --git a/docs/dev/table/python/python_udfs.md b/docs/dev/table/python/python_udfs.md index ea01cd0ca3faba..cb8c4391f71e7c 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 python worker to use the managed memory budget of the task slot +table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) + # register the Python function table_env.register_function("hash_code", udf(HashCode(), DataTypes.BIGINT(), DataTypes.BIGINT())) @@ -76,6 +79,9 @@ public class HashCode extends ScalarFunction { table_env = BatchTableEnvironment.create(env) +# configure the python worker to use the managed memory budget of the task slot +table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) + # register the Java function table_env.register_java_function("hash_code", "my.java.function.HashCode") @@ -145,6 +151,9 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] +# configure the python worker to use the managed memory budget of the task slot +table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) + # register the Python Table Function table_env.register_function("split", udtf(Split(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.INT()])) @@ -182,6 +191,9 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] +# configure the python worker to use the managed memory budget of the task slot +table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) + # Register the java function. table_env.register_java_function("split", "my.java.function.Split") diff --git a/docs/dev/table/python/vectorized_python_udfs.md b/docs/dev/table/python/vectorized_python_udfs.md index 8eec40bf497bc4..f1c40bf95317c3 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 python worker to use the managed memory budget of the task slot +table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) + # register the vectorized Python scalar function table_env.register_function("add", add) From 84411abff8dc30ad529e4fdc0a17636bede5e31d Mon Sep 17 00:00:00 2001 From: "acqua.csq" Date: Fri, 10 Jul 2020 17:42:10 +0800 Subject: [PATCH 2/2] [FLINK-18526][python][docs] Add configuration of Python UDF to use Managed Memory in the doc of Pyflink. --- docs/dev/table/python/python_udfs.md | 24 ++++++++++++------- .../table/python/vectorized_python_udfs.md | 6 +++-- 2 files changed, 20 insertions(+), 10 deletions(-) diff --git a/docs/dev/table/python/python_udfs.md b/docs/dev/table/python/python_udfs.md index cb8c4391f71e7c..58bf528d3452d7 100644 --- a/docs/dev/table/python/python_udfs.md +++ b/docs/dev/table/python/python_udfs.md @@ -48,8 +48,8 @@ class HashCode(ScalarFunction): table_env = BatchTableEnvironment.create(env) -# configure the python worker to use the managed memory budget of the task slot -table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) +# 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())) @@ -61,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 %} @@ -79,8 +81,8 @@ public class HashCode extends ScalarFunction { table_env = BatchTableEnvironment.create(env) -# configure the python worker to use the managed memory budget of the task slot -table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) +# 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") @@ -92,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. @@ -151,8 +155,8 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] -# configure the python worker to use the managed memory budget of the task slot -table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) +# 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()])) @@ -167,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 %} @@ -191,8 +197,8 @@ env = StreamExecutionEnvironment.get_execution_environment() table_env = StreamTableEnvironment.create(env) my_table = ... # type: Table, table schema: [a: String] -# configure the python worker to use the managed memory budget of the task slot -table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) +# 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") @@ -210,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 f1c40bf95317c3..28337f17f4ef94 100644 --- a/docs/dev/table/python/vectorized_python_udfs.md +++ b/docs/dev/table/python/vectorized_python_udfs.md @@ -54,8 +54,8 @@ def add(i, j): table_env = BatchTableEnvironment.create(env) -# configure the python worker to use the managed memory budget of the task slot -table_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True) +# 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) @@ -66,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**.