PyFlink结合AWS Kinesis SQL连接器使用UDTF时遇CAST函数不支持异常

云上老周

2026-8-8

PyFlink UDTF配合Kinesis Connector抛出Unsupported Python SqlFunction CAST异常解决方案

在使用PyFlink Table API读写AWS Kinesis流时，不使用UDTF功能一切正常，但添加带@udtf装饰器的扁平化解构函数后，执行INSERT INTO SQL语句时触发如下异常：

pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Unsupported Python SqlFunction CAST.

核心代码片段：

@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)
table_result = table_env.execute_sql(f"INSERT INTO {output_table_name} SELECT * FROM result_table")

解决方案

消除隐式类型转换
检查输出Kinesis表的Schema，确保UDTF的result_types与表字段类型完全一致，避免Flink触发自动CAST操作。同时移除UDTF函数内的手动类型转换，直接返回与目标类型匹配的原始数据：

@udtf(result_types=[DataTypes.STRING(), DataTypes.INT()])
def flatten_row(row: Row):
    for s in row["members"]:
        # 确保s["id"]为STRING类型，s["name"]为INT类型，与result_types和输出表Schema严格匹配
        yield s["id"], s["name"]

改用Table API原生insert_into方法
避免通过创建临时视图再执行SQL INSERT的方式，直接使用Table API的insert_into方法提交任务，绕开SQL层的类型转换逻辑：
```
result_table = input_table.flat_map(flatten_row).alias("id", "name")
# 直接执行插入操作，无需通过execute_sql
result_table.insert_into(output_table_name).execute()
```
升级Flink版本（可选）
Flink 1.16版本在Python UDTF与SQL交互的类型处理上存在已知兼容性问题，升级到1.17及以上版本可修复部分相关BUG。