（5）FlinkSQL将socket数据写入到mysql方式二

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = tExecutionEnvironment();        tParallelism(1);        StreamTableEnvironment tableEnv = eate(env);        DataStreamSource streamSource = cketTextStream("127.0.0.1", 9999,"\n");        SingleOutputStreamOperator waterDS = (new MapFunction() {            @Override            public WaterSensor map(String s) throws Exception {                String[] split = s.split(",");                return new WaterSensor(split[0], eLong(split[1]), eInt(split[2]));            }        });        // 将流转化为表        Table table = omDataStream(waterDS,                $("id"),                $("ts"),                $("vc"),                $("pt").proctime());        eateTemporaryView("EventTable", table);        tableEnv.executeSql("CREATE TABLE flinksink (" +                "componentname STRING," +                "componentcount BIGINT NOT NULL," +                "componentsum BIGINT" +                ") WITH (" +                "'connector.type' = 'jdbc'," +                "'connector.url' = 'jdbc:mysql://localhost:3306/testdb?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai'," +                "'le' = 'flinksink'," +                "'r' =  'r'," +                "'ername' = 'root'," +                "'ssword' = 'root'," +                "'x-rows'='3'\r\n" +                ")"        );        Table mysql_user = om("flinksink");        mysql_intSchema();        Table result = tableEnv.sqlQuery(                "SELECT " +                        "id as componentname, " +                //window_start, window_end,                        "COUNT(ts) as componentcount ,SUM(ts) as componentsum " +                        "FROM TABLE( " +                        "TUMBLE( TABLE EventTable , " +                        "DESCRIPTOR(pt), " +                        "INTERVAL '10' SECOND)) " +                        "GROUP BY id , window_start, window_end"        );        //方式一：写入数据库//        result.executeInsert("flinksink").print(); //;.insertInto("flinksink");        //方式二：写入数据库        eateTemporaryView("ResultTable", result);        tableEnv.executeSql("insert into flinksink SELECT * FROM ResultTable").print();//        AppendStream(result, ass).print("toAppendStream");           //追加模式        env.execute();    }