Flink 1.15.4创建TableEnvironment报错：inBatchMode不存在于JVM

环境版本：Python 3.6.3，Apache Flink 1.15.4

问题重现 #

按照Flink 1.15官方文档创建批处理TableEnvironment时，触发如下错误：
py4j.protocol.Py4JError: org.apache.flink.table.api.EnvironmentSettings.inBatchMode does not exist in the JVM

代码示例 #

from pyflink.table import EnvironmentSettings, TableEnvironment

# 创建批处理TableEnvironment
env_settings = EnvironmentSettings.in_batch_mode()
table_env = TableEnvironment.create(env_settings)

table = table_env.from_elements([(1, 'Hi'), (2, 'Hello')])
table.to_pandas()

执行报错输出 #

$ python3 demo.py 
Traceback (most recent call last):
  File "demo.py", line 4, in <module>
    env_settings = EnvironmentSettings.in_batch_mode()
  File "/home/venv/pyflink_1.15.4/lib64/python3.6/site-packages/pyflink/table/environment_settings.py", line 238, in in_batch_mode
    get_gateway().jvm.EnvironmentSettings.inBatchMode())
  File "/home/venv/pyflink_1.15.4/lib64/python3.6/site-packages/py4j/java_gateway.py", line 1548, in __getattr__
    "{0}.{1} does not exist in the JVM".format(self._fqn, name))
py4j.protocol.Py4JError: org.apache.flink.table.api.EnvironmentSettings.inBatchMode does not exist in the JVM

问题原因 #

Flink 1.15.x的Java API中并未提供EnvironmentSettings.inBatchMode()静态方法，该方法是Flink 1.16及后续版本才新增的。官方文档的nightlies版本示例混用了新版本API，导致与1.15版本不兼容。

解决方法 #

在Flink 1.15中，需通过EnvironmentSettings.new_instance()构建器来创建批处理环境，修改后的代码如下：

from pyflink.table import EnvironmentSettings, TableEnvironment

# 创建批处理TableEnvironment（适配Flink 1.15）
env_settings = EnvironmentSettings.new_instance().in_batch_mode().build()
table_env = TableEnvironment.create(env_settings)

table = table_env.from_elements([(1, 'Hi'), (2, 'Hello')])
print(table.to_pandas())

该代码通过new_instance()生成构建器实例，调用in_batch_mode()切换为批处理模式，最后用build()生成可用的EnvironmentSettings对象，完全适配Flink 1.15版本的API规范。

内容的提问来源于stack exchange，提问作者Damon Chen