update

Author: Java-Edge

.gitignore

@@ -3,3 +3,6 @@
 *.class
 *.log
 
+/.idea
+target
+.idea

pom.xml

@@ -11,19 +11,19 @@
 	<name>Flink Tutorial</name>
 	<url>http://www.myorganization.org</url>
 
-	<repositories>
-		<repository>
-			<id>apache.snapshots</id>
-			<name>Apache Development Snapshot Repository</name>
-			<url>https://repository.apache.org/content/repositories/snapshots/</url>
-			<releases>
-				<enabled>false</enabled>
-			</releases>
-			<snapshots>
-				<enabled>true</enabled>
-			</snapshots>
-		</repository>
-	</repositories>
+<!--	<repositories>-->
+<!--		<repository>-->
+<!--			<id>apache.snapshots</id>-->
+<!--			<name>Apache Development Snapshot Repository</name>-->
+<!--			<url>https://repository.apache.org/content/repositories/snapshots/</url>-->
+<!--			<releases>-->
+<!--				<enabled>false</enabled>-->
+<!--			</releases>-->
+<!--			<snapshots>-->
+<!--				<enabled>true</enabled>-->
+<!--			</snapshots>-->
+<!--		</repository>-->
+<!--	</repositories>-->
 
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

src/main/java/TestJava.java

@@ -1,5 +1,3 @@
-import org.relaxng.datatype.DatatypeException;
-
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.Date;

src/main/java/com/javaedge/java/chapter3/JavaDataSetDataSourceApp.java

@@ -14,12 +14,12 @@ public class JavaDataSetDataSourceApp {
 
     public static void main(String[] args) throws Exception {
         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-//        fromCollection(env);
+        fromCollection(env);
         textFile(env);
     }
 
     public static void textFile(ExecutionEnvironment env) throws Exception {
-        String filePath = "file:///Volumes/doc/data/data.txt";
+        String filePath = "file:///Volumes/Download/data/data.txt";
         env.readTextFile(filePath).print();
         System.out.println("===========~这是一个分割线~============");
 

src/main/java/com/javaedge/java/chapter4/JavaDataStreamSourceApp.java

@@ -13,8 +13,8 @@ public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
 //      socketFunction(env);
-//      nonParallelSourceFunction(env);
-        parallelSourceFunction(env);
+      nonParallelSourceFunction(env);
+//        parallelSourceFunction(env);
 
         env.execute("JavaDataStreamSourceApp");
     }

src/main/java/com/javaedge/java/chapter4/JavaDataStreamTransformationApp.java

@@ -17,9 +17,9 @@
 public class JavaDataStreamTransformationApp {
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-//      filterFunction(env);
+      filterFunction(env);
 //      unionFunction(env);
-        splitSelectFunction(env);
+//        splitSelectFunction(env);
         env.execute("JavaDataStreamTransformationApp");
     }
 

src/main/java/com/javaedge/java/chapter5/JavaTableSQLAPI.java

@@ -22,7 +22,7 @@ public static void main(String[] args) throws Exception {
 
         DataSet<Sales> csv = env.readCsvFile(filePath)
                 .ignoreFirstLine()
-                .pojoType(Sales.class,"transactionId","customerId","itemId","amountPaid");
+                .pojoType(Sales.class, "transactionId", "customerId", "itemId", "amountPaid");
         //csv.print();
 
         Table sales = tableEnv.fromDataSet(csv);
@@ -33,11 +33,11 @@ public static void main(String[] args) throws Exception {
         result.print();
     }
 
-    public static class Sales{
+    public static class Sales {
         public String transactionId;
         public String customerId;
         public String itemId;
-        public Double  amountPaid;
+        public Double amountPaid;
     }
 }
 

src/main/java/com/javaedge/java/chapter6/JavaWindowsApp.java

@@ -9,25 +9,24 @@
 
 /**
  * @author JavaEdge
- *
  * @date 2019-07-23
  */
 public class JavaWindowsApp {
 
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        DataStreamSource<String> text = env.socketTextStream("localhost",9999);
-        text.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
-            @Override
-            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
-                String[] tokens = value.toLowerCase().split(",");
-                for(String token : tokens) {
-                    if(token.length() > 0) {
-                        out.collect(new Tuple2<>(token, 1));
+        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);
+        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
+                    @Override
+                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
+                        String[] tokens = value.toLowerCase().split(",");
+                        for (String token : tokens) {
+                            if (token.length() > 0) {
+                                out.collect(new Tuple2<>(token, 1));
+                            }
+                        }
                     }
-                }
-            }
-        }).keyBy(0)
+                }).keyBy(0)
                 .timeWindow(Time.seconds(5))
                 .sum(1)
                 .print()