Error with

key_by

and

reduce

operators when locally running pyflink 1.13. Here is the code:

env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)
    ds = env.from_collection(
        collection=[(1, 2), (2, 3), (1, 3), (2, 4)],
        type_info=Types.ROW([<http://Types.INT|Types.INT>(), <http://Types.INT|Types.INT>()]))
    concat_stream = ds.key_by(lambda x: x[0], key_type=<http://Types.INT|Types.INT>()).reduce(lambda a, b: (a[0], a[1] + b[1]))
    concat_stream.print()
    env.execute("tutorial_job")

This is the key error message:

AttributeError: 'tuple' object has no attribute 'get_fields_by_names'

is there anything wrong with the code?

@Alessio Marinelli I guess you could try to change

Types.ROW

to

Types.TUPLE

as the data is indeed tuple instead of row.

💪

@Alessio Marinelli Oh, sorry.😂 cc @mralfredgui

no problem

Thanks a lot and it works.