如何将Table API与SQL结合使用？

DataStream<Tuple2<String, Integer>> inputDS = env.addSource(...);
// 通过Table API注册为临时表
Table inputTable = tableEnv.fromDataStream(inputDS, "user, cnt");
tableEnv.createTemporaryView("input_table", inputTable);
// 通过SQL查询
Table sqlResult = tableEnv.sqlQuery(
    "SELECT user, SUM(cnt) as total FROM input_table GROUP BY user"
);

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 通过SQL创建源表
tableEnv.executeSql(
    "CREATE TABLE kafka_source (" +
    "  user STRING, " +
    "  cnt INT" +
    ") WITH ('connector' = 'kafka', ...)"
);
// 通过Table API查询并过滤数据
Table resultTable = tableEnv.from("kafka_source")
    .filter("cnt > 100")
    .select("user, cnt");
// 转换为DataStream
DataStream<Row> resultDS = tableEnv.toAppendStream(resultTable, Row.class);