FlinkSQL与DataStream API如何互操作？

// 定义SQL表（例如Kafka源）
tableEnv.executeSql(
    "CREATE TABLE kafka_source (" +
    "  user STRING, " +
    "  cnt INT" +
    ") WITH (...)"
);
// 转换为DataStream
Table table = tableEnv.sqlQuery("SELECT * FROM kafka_source");
DataStream<Tuple2<String, Integer>> ds = tableEnv.toAppendStream(table, Row.class);

// SQL预处理数据（过滤）
Table filteredTable = tableEnv.sqlQuery("SELECT * FROM kafka_source WHERE cnt > 100");
DataStream<Row> filteredDS = tableEnv.toAppendStream(filteredTable);
// 使用DataStream API做窗口聚合
DataStream<Tuple2<String, Integer>> windowedDS = filteredDS
    .map(row -> Tuple2.of(row.getField(0), row.getField(1)))
    .keyBy(value -> value.f0)
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .sum(1);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 创建DataStream（例如Kafka数据）
DataStream<Tuple2<String, Integer>> inputDS = env.addSource(...);
// 注册为表（需定义Schema）
tableEnv.createTemporaryView("kafka_source", inputDS, 
    Schema.newBuilder()
        .column("user", DataTypes.STRING())
        .column("cnt", DataTypes.INT())
        .build());

-- SQL查询并过滤数据
SELECT user, SUM(cnt) as total 
FROM kafka_source 
GROUP BY user;

// 执行SQL并获取结果流
TableResult result = tableEnv.executeSql(
    "SELECT user, SUM(cnt) as total FROM kafka_source GROUP BY user");
// 转换为DataStream（需指定类型）
DataStream<Row> resultDS = tableEnv.toAppendStream(result, Row.class);
resultDS.print();