perf:优化

Java-Edge · Java-Edge · commit fc674377f448 · 2024-08-02T13:16:02.000+08:00
diff --git a/.idea/encodings.xml b/.idea/encodings.xml
diff --git a/README.md b/README.md
@@ -1,16 +1,16 @@
 # Flink 从入门到实战 - Scala/Java双语言版本
 
-# 1 核心知识点
+## 1 核心知识点
 ![](https://upload-images.jianshu.io/upload_images/16782311-dc4156dc0a34d557.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## 编程模型及核心概念
+### 编程模型及核心概念
 Flink中的DataSet & DataStream
 
 Flink编程模型、延迟执行
 
 Flink中支持的数据类型
 
 ![](https://upload-images.jianshu.io/upload_images/16782311-77ad64efc672f89b.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## DataSet API编程
+### DataSet API编程
 
 DataSet中的Data Source、Sink
 
@@ -20,7 +20,7 @@ Transformation核心应用
 
   
 ![](https://upload-images.jianshu.io/upload_images/16782311-6d7c1b11f42bb125.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## DataStream API编程
+### DataStream API编程
 
 DataStream中的Data Source及自定义实现
 
@@ -31,7 +31,7 @@ Transformation核心应用
 
 
 ![](https://upload-images.jianshu.io/upload_images/16782311-8ec74ae8f1b686da.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## Flink Table API&SQL编程
+### Flink Table API&SQL编程
 
 Flink关系型API介绍
 
@@ -41,7 +41,7 @@ Table&SQL API其他功能
  
 
 ![](https://upload-images.jianshu.io/upload_images/16782311-7f673e9fb901a151.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## Flink中的Time及Windows操作
+### Flink中的Time及Windows操作
 
 三种不同Time详解
 
@@ -79,11 +79,8 @@ Flink常见的监控方式
 
 常见的Flink调优策略
 
-# 2 以下知名公司都在使用Flink
+# 2 公司都在用Flink
 ![](https://upload-images.jianshu.io/upload_images/16782311-9447a7c96178832b.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
 
-# X 交流学习
-![](https://upload-images.jianshu.io/upload_images/16782311-8d7acde57fdce062.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-## [Java交流群](https://jq.qq.com/?_wv=1027&k=5UB4P1T)
-## [博客](http://www.shishusheng.com)
-## [Github](https://github.com/Wasabi1234)
+## X 交流学习
+### [Github](https://github.com/Java-Edge)
diff --git a/pom.xml b/pom.xml
@@ -311,4 +311,4 @@
 		</profile>
 	</profiles>
 
-</project>
+</project>
diff --git a/src/main/java/com/javaedge/java/chapter3/JavaDataSetSinkApp.java b/src/main/java/com/javaedge/java/chapter3/JavaDataSetSinkApp.java
@@ -1,7 +1,7 @@
 package com.javaedge.java.chapter3;
 
 /**
- * @author sss
+ * @author JavaEdge
  * @date 2019-07-18
  */
 
diff --git a/src/main/java/com/javaedge/java/chapter4/JavaCustomSinkToMySQL.java b/src/main/java/com/javaedge/java/chapter4/JavaCustomSinkToMySQL.java
@@ -5,11 +5,6 @@
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
-/**
- * @author JavaEdge
- *
- * @date 2019-07-21
- */
 public class JavaCustomSinkToMySQL {
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/src/main/java/com/javaedge/java/chapter4/SinkToMySQL.java b/src/main/java/com/javaedge/java/chapter4/SinkToMySQL.java
@@ -7,12 +7,7 @@
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 
-/**
- * @author JavaEdge
- *
- * @date 2019-07-21
- */
-public class SinkToMySQL extends RichSinkFunction<Student>{
+public class SinkToMySQL extends RichSinkFunction<Student> {
 
     Connection connection;
     PreparedStatement pstmt;
@@ -25,7 +20,7 @@ private Connection getConnection() {
 
             String url = "jdbc:mysql://localhost:3306/javaedge_flink";
 
-            conn = DriverManager.getConnection(url,"root","root");
+            conn = DriverManager.getConnection(url, "root", "123456");
 
         } catch (Exception e) {
             e.printStackTrace();
@@ -36,6 +31,7 @@ private Connection getConnection() {
 
     /**
      * 在open方法中建立connection
+     *
      * @param parameters
      * @throws Exception
      */
@@ -73,17 +69,18 @@ public void invoke(Student value, Context context) throws Exception {
 
     /**
      * 在close方法中要释放资源
+     *
      * @throws Exception
      */
     @Override
     public void close() throws Exception {
         super.close();
 
-        if(pstmt != null) {
+        if (pstmt != null) {
             pstmt.close();
         }
 
-        if(connection != null) {
+        if (connection != null) {
             connection.close();
         }
     }
diff --git a/src/main/java/com/javaedge/java/chapter4/Student.java b/src/main/java/com/javaedge/java/chapter4/Student.java
@@ -14,5 +14,4 @@ public class Student {
     private String name;
 
     private int age;
-}
-
+}
diff --git a/src/main/java/com/javaedge/java/chapter6/JavaWindowsProcessApp.java b/src/main/java/com/javaedge/java/chapter6/JavaWindowsProcessApp.java
@@ -10,28 +10,23 @@
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 
-/**
- * @author JavaEdge
- *
- * @date 2019-07-23
- */
 public class JavaWindowsProcessApp {
 
     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<Integer, Integer>>() {
-            @Override
-            public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
-                String[] tokens = value.toLowerCase().split(",");
-                for (String token : tokens) {
-                    if (token.length() > 0) {
-                        out.collect(new Tuple2<>(1, Integer.parseInt(token)));
+                    @Override
+                    public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
+                        String[] tokens = value.toLowerCase().split(",");
+                        for (String token : tokens) {
+                            if (token.length() > 0) {
+                                out.collect(new Tuple2<>(1, Integer.parseInt(token)));
+                            }
+                        }
                     }
-                }
-            }
-        }).keyBy(0)
+                }).keyBy(0)
                 .timeWindow(Time.seconds(5))
                 .process(new ProcessWindowFunction<Tuple2<Integer, Integer>, Object, Tuple, TimeWindow>() {
                     @Override
diff --git a/src/main/java/com/javaedge/java/chapter6/JavaWindowsReduceApp.java b/src/main/java/com/javaedge/java/chapter6/JavaWindowsReduceApp.java
@@ -8,28 +8,23 @@
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.util.Collector;
 
-/**
- * @author JavaEdge
- *
- * @date 2019-07-23
- */
 public class JavaWindowsReduceApp {
 
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        DataStreamSource<String> text = env.socketTextStream("localhost",9999);
+        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);
 
-        text.flatMap(new FlatMapFunction<String, Tuple2<Integer,Integer>>() {
-            @Override
-            public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
-                String[] tokens = value.toLowerCase().split(",");
-                for(String token : tokens) {
-                    if(token.length() > 0) {
-                        out.collect(new Tuple2<>(1, Integer.parseInt(token)));
+        text.flatMap(new FlatMapFunction<String, Tuple2<Integer, Integer>>() {
+                    @Override
+                    public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
+                        String[] tokens = value.toLowerCase().split(",");
+                        for (String token : tokens) {
+                            if (token.length() > 0) {
+                                out.collect(new Tuple2<>(1, Integer.parseInt(token)));
+                            }
+                        }
                     }
-                }
-            }
-        }).keyBy(0)
+                }).keyBy(0)
                 .timeWindow(Time.seconds(5))
                 .reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
                     @Override
diff --git a/src/main/resources/init.sql b/src/main/resources/init.sql
@@ -0,0 +1,9 @@
+create database javaedge_flink;
+
+create table student
+(
+    id   int(11) NOT NULL AUTO_INCREMENT,
+    name varchar(25),
+    age  int(10),
+    primary key (id)
+);
diff --git a/src/main/scala/com/javaedge/scala/chapter3/DBUtils.scala b/src/main/scala/com/javaedge/scala/chapter3/DBUtils.scala
@@ -3,7 +3,7 @@ package com.javaedge.scala.chapter3
 import scala.util.Random
 
 /**
-  * @author sss
+  * @author JavaEdge
   * @date 2019-07-17
   *
   */
diff --git a/src/main/scala/com/javaedge/scala/chapter4/CustomNonParallelSourceFunction.scala b/src/main/scala/com/javaedge/scala/chapter4/CustomNonParallelSourceFunction.scala
@@ -2,13 +2,6 @@ package com.javaedge.scala.chapter4
 
 import org.apache.flink.streaming.api.functions.source.SourceFunction
 
-/**
-  * 自定义
-  *
-  * @author JavaEdge
-  * @date 2019-07-19
-  *
-  */
 class CustomNonParallelSourceFunction extends SourceFunction[Long]{
 
   var count = 1L
diff --git a/src/main/scala/com/javaedge/scala/chapter6/WindowsReduceApp.scala b/src/main/scala/com/javaedge/scala/chapter6/WindowsReduceApp.scala
@@ -3,11 +3,7 @@ package com.javaedge.scala.chapter6
 import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
 import org.apache.flink.streaming.api.windowing.time.Time
 import org.apache.flink.api.scala._
-/**
-  * @author JavaEdge
-  *
-  * @date 2019-07-23
-  */
+
 object WindowsReduceApp {
 
   def main(args: Array[String]): Unit = {
@@ -28,7 +24,4 @@ object WindowsReduceApp {
 
     env.execute("WindowsReduceApp")
   }
-}
-
-
-
+}
diff --git a/src/main/scala/com/javaedge/scala/chapter7/FileSystemSinkApp.scala b/src/main/scala/com/javaedge/scala/chapter7/FileSystemSinkApp.scala
@@ -4,10 +4,6 @@ import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
 import org.apache.flink.streaming.connectors.fs.StringWriter
 import org.apache.flink.streaming.connectors.fs.bucketing.{BucketingSink, DateTimeBucketer}
 
-/**
-  * @author JavaEdge
-  * @date 2019-07-23
-  */
 object FileSystemSinkApp {
 
   def main(args: Array[String]): Unit = {
