sync

maven/.gitignore

@@ -106,3 +106,4 @@ buildNumber.properties
 .mvn/timing.properties
 .mvn/wrapper/maven-wrapper.jar
 
+/.idea/

maven/src/main/java/com/duanledexianxianxian/maven/flink/sql/sql/Main.java

@@ -7,7 +7,10 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
 import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
 import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Over;
 import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.types.Row;
 
@@ -35,8 +38,12 @@ public class Main {
         Table table = tableEnv.fromDataStream(source, "eventTime.rowtime");
 
         // register DataStream as Table
-        tableEnv.registerTable("Zhisheng", table);
-        Table sorted = tableEnv.sqlQuery("select * from Zhisheng ");
+//        tableEnv.registerTable("Zhisheng", table);
+        Table sorted = table.window(Over.partitionBy("eventTime")
+                .orderBy("eventTime.desc")
+                .preceding("20.second")
+                .following("CURRENT_RANGE")
+                .as("window")).select("eventTime ");
         DataStream<Row> rowDataStream = tableEnv.toAppendStream(sorted, Row.class);
 
         rowDataStream.print();
@@ -80,7 +87,7 @@ public class Main {
         public void run(SourceContext<Event> ctx) throws Exception {
             while (running) {
                 ctx.collect(new Event());
-                Thread.sleep(1);
+                Thread.sleep(5000);
             }
         }