V1.13.2版本Python API是否支持DataStream和Table的相互转换

Flink 1.13.2版本的Python API完全支持DataStream和Table两类数据结构的互相转换，转换操作通过StreamTableEnvironment提供的标准接口实现，具体用法如下：

DataStream转Table #

调用StreamTableEnvironment的from_data_stream()方法即可完成转换，支持主动指定列名、时间属性等配置：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.common import Row, Types

# 初始化运行环境
ds_env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(ds_env)

# 构造测试DataStream
source_ds = ds_env.from_collection(
    [Row(1, "Alice", 25), Row(2, "Bob", 30), Row(3, "Cindy", 28)],
    type_info=Types.ROW([Types.INT(), Types.STRING(), Types.INT()])
)

# DataStream转Table，指定列名
source_table = t_env.from_data_stream(source_ds).alias("id", "name", "age")

注意：转换过程会自动适配Flink类型系统，基础类型、Row类型、自定义POJO类型均可直接兼容。

Table转DataStream #

需要根据Table的查询逻辑选择对应转换方式：

• 若Table仅输出新增数据（无聚合、关联更新等操作），可直接用to_data_stream()转普通追加流
• 若Table输出包含更新、撤回数据（如分组聚合、upsert查询场景），需用to_retract_stream()转回撤流，流内每条数据会携带布尔标记位，标记为True表示新增记录，False表示撤回记录

追加流转换示例 #

# 仅做过滤映射，无更新逻辑，可转追加流
filter_table = t_env.sql_query("SELECT id, name FROM %s WHERE age > 26" % source_table)
result_ds = t_env.to_data_stream(filter_table)
result_ds.print()

ds_env.execute()

回撤流转换示例 #

# 带分组聚合，会产生更新数据，需转回撤流
agg_table = t_env.sql_query("SELECT age, COUNT(*) as user_count FROM %s GROUP BY age" % source_table)
# 第二个参数指定返回数据类型，结构为<回撤标记, 数据行>
retract_ds = t_env.to_retract_stream(
    agg_table,
    Types.ROW([Types.BOOLEAN(), Types.ROW([Types.INT(), Types.BIGINT()])])
)
retract_ds.print()

ds_env.execute()

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