dianfu commented on a change in pull request #12370:
URL: https://github.com/apache/flink/pull/12370#discussion_r431679353
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonPythonBase.scala
##########
@@ -150,6 +154,35 @@ trait CommonPythonBase {
}
realEnv
}
+
+ private def usingManagedMemory(config: Configuration): Boolean = {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ config.getBoolean(clazz.getField("USE_MANAGED_MEMORY").get(null)
+ .asInstanceOf[ConfigOption[java.lang.Boolean]])
+ }
+
+ private def getPythonWorkerMemory(config: Configuration): Long = {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ val pythonFrameworkMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_FRAMEWORK_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ val pythonBufferMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_DATA_BUFFER_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ pythonFrameworkMemorySize.add(pythonBufferMemorySize).getBytes
+ }
+
+ private def setDefaultConfigurations(config: Configuration): Unit = {
+ if (!usingManagedMemory(config) &&
!config.contains(TaskManagerOptions.TASK_OFF_HEAP_MEMORY)) {
+ // Set the default value for the task off-heap memory if not set if the
Python worker isn't
+ // using managed memory. Note that this is best effort attempt and if a
task slot contains
+ // multiple Python workers, users should set the task off-heap memory
manually.
+ config.setString(TaskManagerOptions.TASK_OFF_HEAP_MEMORY.key(),
+ getPythonWorkerMemory(config) + "b")
+ }
Review comment:
It's a best effort attempt and should work in most scenarios. It's only
activated if managed memory is disabled for Python UDF and users have not
config the task off-heap memory. In this case, it doesn't no harm to set a
default task off-heap memory for Python UDF.
Besides, For Python users, ease of use is the most important thing.
Configuring the task off-heap memory is not an easy thing and so we want to let
users to think about that as little as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
