pyflink Unsupported Python SqlFunction CAST when working with amazon-kinesis-sql-connector and udtf function

Question

i am currently trying to get Pyflink running with the AWS-Kinesis-SQL-Connector.

A use the TableAPI and can read from Kinesis and also write back to another Kinesis Stream. As soon as i use a udtf decorated function i get the following exception:

  File "/home/user/anaconda3/envs/flink-env/lib/python3.8/site-packages/pyflink/table/table_environment.py", line 828, in execute_sql
    return TableResult(self._j_tenv.executeSql(stmt))
  File "/home/user/anaconda3/envs/flink-env/lib/python3.8/site-packages/py4j/java_gateway.py", line 1321, in __call__
    return_value = get_return_value(
  File "/home/user/anaconda3/envs/flink-env/lib/python3.8/site-packages/pyflink/util/exceptions.py", line 158, in deco
    raise java_exception
pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Unsupported Python SqlFunction CAST.

I try to sum up the core sn-ps of the script:

@udtf(result_types=[DataTypes.STRING(), DataTypes.INT()])
def flatten_row(row: Row) -> Row:
   for s in row["members"]:
       yield Row(str(s["id"]), s["name"])

result_table = input_table.flat_map(flatten_row).alias("id", "name")
table_env.create_temporary_view("result_table", result_table)

As soon as i want to execute it on the Stream the exception get's raised.

table_result = table_env.execute_sql(f"INSERT INTO {output_table_name} SELECT * FROM result_table")

The output_table and input_table are connected to Kinesis Streams and without the udtf function it works.

Environment

Used apache-flink==1.16.0 and python3.8. Tried Conda and PIP environments

Thank you!

Already tried different versions of the apache-flink and the amazon-kinesis-sql-connector. Conda and PIP environments with Python3.8.

Answer 1

Finally i found out that the problem was the JDK version pre-installed on my MacOS. I downgraded from 15.0.2 until i reached 11.0.16, which was finally working without any error. So it seems that the Python apache-flink package needs older JDK versions.