flink相关回撤流的信息

r囧r小猫 2022-11-17 13:54 144阅读 0赞

1、回撤流的信息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70][]

2、流中包含的数据

StreamTableEnvironment

3、sql代码调试的情况

package architecture;
    
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    /**
     * 项目名称: Apache Flink 知其然，知其所以然 - architecture
     * 功能描述: 从Kafka读取消息数据，并在控制台进行打印。
     * 操作步骤: 1. 直接执行作业，看到输出
     * 2. 调试查看sql的plan
     * 3. 断点，查看codegen 代码
     * 作者： 孙金城
     * 日期： 2020/10/1
     */
    public class DataGen2Print {
        public static void main(String[] args) throws Exception {
            // DataGen
            String sourceDDL = "CREATE TABLE stu (\n" +
                    " name STRING , \n" +
                    " age INT \n" +
                    ") WITH (\n" +
                    " 'connector' = 'datagen' ,\n" +
                    " 'rows-per-second'='5',\n"+
                    " 'fields.name.length'='10'\n" +
                    ")";
            // Print
            String sinkDDL = "CREATE TABLE sink_table (\n" +
                    " name STRING , \n" +
                    " age INT , \n" +
                    " weight INT  \n" +
                    ") WITH (\n" +
                    " 'connector' = 'print'\n" +
                    ")";
            // 创建执行环境
            EnvironmentSettings settings = EnvironmentSettings
                    .newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
            StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(sEnv, settings);
    
            //注册source和sink
            tEnv.executeSql(sourceDDL);
            tEnv.executeSql(sinkDDL);
    
            String sql = "INSERT INTO sink_table SELECT stu.name, stu.age-2, 35+2 FROM stu";
    
            String plan = tEnv.explainSql(sql);
            System.out.println(plan);
    
            tEnv.executeSql(sql);
        }
    }

4、flinksql的规则

FlinkStreamRuleSets

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70]: /images/20221022/a776914350dd4a8896fa8c90c7545fa4.png