Flink Table API如何转换为DataStream或DataSet？

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
// 创建表并执行查询
Table resultTable = tableEnv.sqlQuery("SELECT name, age FROM users WHERE age > 30");
// 转换为DataSet
DataSet<Row> dataSet = tableEnv.toDataSet(resultTable, Row.class);
// 处理DataSet
dataSet.print();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 创建表并执行查询
Table resultTable = tableEnv.sqlQuery("SELECT user, COUNT(url) FROM clicks GROUP BY user");
// 转换为DataStream（根据表是否更新选择方法）
// 场景1：仅插入结果（如无GROUP BY的过滤）
DataStream<Row> appendStream = tableEnv.toAppendStream(resultTable, Row.class);
// 场景2：含更新操作（如GROUP BY聚合）
DataStream<Tuple2<Boolean, Row>> retractStream = 
    tableEnv.toRetractStream(resultTable, Row.class);
// 进一步处理DataStream
retractStream.print();
env.execute();