Hello team, I’m using PyFlink and found the

DataStream.map()

in python seems doesn’t work. I write a Python demo like:

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env.enable_checkpointing(1000)

source = KafkaSource.builder().build()

stream = env.from_source(source, WatermarkStrategy.no_watermarks(), "team_config_source")

sink = FileSink \
    .for_row_format('/opt/result/', Encoder.simple_string_encoder("UTF-8")) \
    .with_output_file_config(OutputFileConfig.builder()
                             .with_part_prefix("team_config")
                             .with_part_suffix(".json")
                             .build()) \
    .with_rolling_policy(RollingPolicy.default_rolling_policy(part_size=1024 ** 3, rollover_interval=15 * 60 * 1000,
                                                              inactivity_interval=5 * 60 * 1000)) \
    .build()

def mapping(data):
    return data

stream.map(mapping, BasicTypeInfo.STRING_TYPE_INFO()).sink_to(sink)
env.execute()

But Flink gives me this exception:

2023-01-18 11:34:34 Traceback (most recent call last):
2023-01-18 11:34:34   File "/usr/local/lib/python3.8/runpy.py", line 194, in _run_module_as_main
2023-01-18 11:34:34     return _run_code(code, main_globals, None,
2023-01-18 11:34:34   File "/usr/local/lib/python3.8/runpy.py", line 87, in _run_code
2023-01-18 11:34:34     exec(code, run_globals)
2023-01-18 11:34:34   File "/tmp/pyflink/00606d52-b6c1-4e13-b7cb-73ee8e196db6/42be79fb-c8bb-4de1-b0fb-c89a7702cddc/flink_driver.py", line 223, in <module>
2023-01-18 11:34:34     process2()
2023-01-18 11:34:34   File "/tmp/pyflink/00606d52-b6c1-4e13-b7cb-73ee8e196db6/42be79fb-c8bb-4de1-b0fb-c89a7702cddc/flink_driver.py", line 218, in process2
2023-01-18 11:34:34     stream.map(mapping, BasicTypeInfo.STRING_TYPE_INFO()).sink_to(sink)
2023-01-18 11:34:34   File "/opt/flink/opt/python/pyflink.zip/pyflink/datastream/data_stream.py", line 312, in map
2023-01-18 11:34:34   File "/opt/flink/opt/python/pyflink.zip/pyflink/datastream/data_stream.py", line 654, in process
2023-01-18 11:34:34   File "<frozen importlib._bootstrap>", line 991, in _find_and_load
2023-01-18 11:34:34   File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
2023-01-18 11:34:34   File "<frozen importlib._bootstrap>", line 655, in _load_unlocked
2023-01-18 11:34:34   File "<frozen importlib._bootstrap>", line 618, in _load_backward_compatible
2023-01-18 11:34:34   File "<frozen zipimport>", line 259, in load_module
2023-01-18 11:34:34   File "/opt/flink/opt/python/pyflink.zip/pyflink/fn_execution/flink_fn_execution_pb2.py", line 38, in <module>
2023-01-18 11:34:34 AttributeError: 'NoneType' object has no attribute 'message_types_by_name'

According to the exception, I look into the

flink_fn_execution_pb2.py

, and I found the code in this file’s beginning was:

DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile('xxx')
_INPUT = DESCRIPTOR.message_types_by_name['Input']

But seems

DescriptorPool.AddSerializedFile()

doesn’t have a return statement, so

DESCRIPTOR

is always

None

. Am I using the

DataStream.map()

in a wrong way or there is something wrong? My environment is: • Flink 1.16.0 • Python Package: apache-flink==1.16.0

@Zeyu Qiu This issue may be similar to this JIRA: https://issues.apache.org/jira/browse/FLINK-28786. It will be fixed in 1.16.1 which is planned to be released soon (already under vote). For now, you could try if that fixes works using the RC package: https://dist.apache.org/repos/dist/dev/flink/flink-1.16.1-rc1/python/ pip install apache-flink-libraries-1.16.1.tar.gz pip install apache_flink-1.16.1-cp38-xxx.whl