/**
 * FlinkQuerySqlDemo 类用于演示如何使用 Flink 执行 SQL 查询
 */
public class FlinkQuerySqlDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置执行环境的并行度为1，便于调试和理解程序
        env.setParallelism(1);
        // 创建TableEnvironment以支持SQL操作
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 定义一个DDL语句来创建一个表user，该表的数据来源于文件系统中的input.csv文件，格式为CSV
        String createDDL = "CREATE TABLE user_event (" +
                " name STRING, " +
                " age INT, " +
                " idn STRING, " +
                " crtTime STRING" +
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input.csv', " +
                " 'format' =  'csv' " +
                ")";

        // 执行DDL语句，实际创建表user
        tableEnv.executeSql(createDDL);

        // 执行一个SQL查询，对user按idn分组计数
        Table result = tableEnv.sqlQuery("select idn ,count(1) as cnt from user_event group by idn");

        // 定义另一个DDL语句，创建一个输出表console，用于将结果数据打印到控制台
        String createPrintOutDDL = "CREATE TABLE user_console (" +
                " idn STRING, " +
                " cnt BIGINT" +
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";

        // 执行创建输出表的DDL
        tableEnv.executeSql(createPrintOutDDL);

        // 将之前查询的结果插入到输出表console中，实现打印输出
        result.executeInsert("user_console");
        // 注释掉了env.execute()，因为在当前上下文中，tableEnv已经执行了executeSql和executeInsert操作，
        // 这些操作内部会触发流处理任务的执行。因此，通常不需要再单独调用env.execute()。
        env.execute();
    }
}