执行pyflink报错 py4j.protocol.Py4JError

[JTjeff]

py4j.protocol.Py4JError: org.apache.flink.streaming.api.utils.PythonTypeUtils.getCollectionInputFormat does not exist in the JVM

我不知道这种报错到底什么意思，初学者，py文件如下：
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment

def data_stream_api_demo():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
ds = env.from_collection(collection=[(1, 'aaa|bb'), (2, 'bb|a'), (3, 'aaa|a')],type_info=Types.ROW([Types.INT(), Types.STRING()]))
def split(s):
splits = s[1].split("|")
for sp in splits:
yield s[0], sp
ds = ds.map(lambda i: (i[0] + 1, i[1])).flat_map(split).key_by(lambda i: i[1]).reduce(lambda i, j: (i[0] + j[0], i[1]))
ds.print()
env.execute()

if name == 'main':
data_stream_api_demo()

应该是我使用的有问题