PyFlink使用execute_and_collect转换SQL Timestamp失败求助

这是PyFlink处理SQL_TIMESTAMP类型时，execute_and_collect反序列化环节的已知问题，并非你的操作错误。

问题原因 #

当你用Types.SQL_TIMESTAMP()定义Row字段类型时，Flink会将Python的datetime对象序列化为Java Timestamp对应的二进制格式，但execute_and_collect在从Java端拉取数据反序列化时，没有正确将该二进制转换回Python的datetime对象，而是直接返回了bytearray，导致后续调用timestamp()方法触发AttributeError。而JDBC sink是通过Java端逻辑直接处理SQL_TIMESTAMP类型，能完成正确的序列化转换，因此可以正常插入数据库。

解决办法 #

1. 改用Types.PYTHON_OBJECT()定义字段类型

这种方式让Flink直接序列化Python对象本身，反序列化时能正确还原成datetime，适合测试场景，但会带来额外的序列化开销，不建议生产大规模使用。修改后的代码：

from datetime import datetime

from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment

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

field_names = ("created_at",)
collection = [(datetime.now(),)]
# 替换为PYTHON_OBJECT类型
types = Types.ROW_NAMED(field_names=field_names, field_types=[Types.PYTHON_OBJECT()])
stream = env.from_collection(collection=collection, type_info=types)

items = stream.execute_and_collect()
print(list(items))
items.close()

2. 转换为可正确序列化的类型（如字符串）

在数据流中将timestamp转换为字符串格式，拉取到客户端后再转回datetime，这种方式性能更稳定，适合生产测试场景：

from datetime import datetime

from pyflink.common import Row
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment

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

# 数据源输出ISO格式的时间字符串
collection = [(datetime.now().isoformat(),)]
field_names = ("created_at",)
field_types = [Types.STRING()]
types = Types.ROW_NAMED(field_names=field_names, field_types=field_types)
stream = env.from_collection(collection=collection, type_info=types)

# 客户端拉取后转换回datetime对象
items = stream.execute_and_collect()
result = [Row(created_at=datetime.fromisoformat(item.created_at)) for item in items]
print(result)
items.close()

3. 升级PyFlink版本

该反序列化问题在PyFlink 1.16及以上版本中已被修复，若你使用的是旧版本，可尝试升级到最新稳定版解决问题。

内容的提问来源于stack exchange，提问作者Felix