update the project version

Author: zhisheng17

flink-learning-common/pom.xml

@@ -54,7 +54,7 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
-            <version>1.9.0</version>
+            <version>${flink.version}</version>
         </dependency>
     </dependencies>
 </project>

flink-learning-common/src/main/java/com/zhisheng/common/utils/ExecutionEnvUtil.java

@@ -7,8 +7,6 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * blog：http://www.54tianzhisheng.cn/
@@ -19,8 +17,7 @@ public static ParameterTool createParameterTool(final String[] args) throws Exce
         return ParameterTool
                 .fromPropertiesFile(ExecutionEnvUtil.class.getResourceAsStream(PropertiesConstants.PROPERTIES_FILE_NAME))
                 .mergeWith(ParameterTool.fromArgs(args))
-                .mergeWith(ParameterTool.fromSystemProperties())
-                .mergeWith(ParameterTool.fromMap(getenv()));
+                .mergeWith(ParameterTool.fromSystemProperties());
     }
 
     public static final ParameterTool PARAMETER_TOOL = createParameterTool();
@@ -29,32 +26,23 @@ private static ParameterTool createParameterTool() {
         try {
             return ParameterTool
                     .fromPropertiesFile(ExecutionEnvUtil.class.getResourceAsStream(PropertiesConstants.PROPERTIES_FILE_NAME))
-                    .mergeWith(ParameterTool.fromSystemProperties())
-                    .mergeWith(ParameterTool.fromMap(getenv()));
+                    .mergeWith(ParameterTool.fromSystemProperties());
         } catch (IOException e) {
             e.printStackTrace();
         }
-        return null;
+        return ParameterTool.fromSystemProperties();
     }
 
     public static StreamExecutionEnvironment prepare(ParameterTool parameterTool) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(parameterTool.getInt(PropertiesConstants.STREAM_PARALLELISM, 5));
         env.getConfig().disableSysoutLogging();
-        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
+        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 60000));
         if (parameterTool.getBoolean(PropertiesConstants.STREAM_CHECKPOINT_ENABLE, true)) {
-            env.enableCheckpointing(parameterTool.getInt(PropertiesConstants.STREAM_CHECKPOINT_INTERVAL, 1000)); // create a checkpoint every 5 seconds
+            env.enableCheckpointing(parameterTool.getInt(PropertiesConstants.STREAM_CHECKPOINT_INTERVAL, 10000));
         }
-        env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
+        env.getConfig().setGlobalJobParameters(parameterTool);
         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
         return env;
     }
-
-    private static Map<String, String> getenv() {
-        Map<String, String> map = new HashMap<>();
-        for (Map.Entry<String, String> entry : System.getenv().entrySet()) {
-            map.put(entry.getKey(), entry.getValue());
-        }
-        return map;
-    }
-}
+}

flink-learning-common/src/main/java/com/zhisheng/common/utils/HttpUtil.java

@@ -29,6 +29,33 @@ public class HttpUtil {
 
     private static final CloseableHttpClient httpClient = HttpClients.createDefault();
 
+    public static String doPostString(String url, String jsonParams) throws Exception {
+        CloseableHttpResponse response = null;
+        HttpPost httpPost = new HttpPost(url);
+
+        String httpStr;
+        try {
+            StringEntity entity = new StringEntity(jsonParams, "UTF-8");
+            entity.setContentEncoding("UTF-8");
+            entity.setContentType("application/json");
+
+            httpPost.setEntity(entity);
+            httpPost.setHeader("content-type", "application/json");
+            //如果要设置 Basic Auth 的话
+//        httpPost.setHeader("Authorization", getHeader());
+            response = httpClient.execute(httpPost);
+            httpStr = EntityUtils.toString(response.getEntity(), "UTF-8");
+
+        } finally {
+            if (response != null) {
+                EntityUtils.consume(response.getEntity());
+                response.close();
+            }
+        }
+        return httpStr;
+    }
+
+
     /**
      * 通过GET方式发起http请求
      */
@@ -60,52 +87,21 @@ public static String doGet(String url) {
     }
 
 
-    /**
-     * 发送 POST 请求（HTTP），JSON形式
-     *
-     * @param url        调用的地址
-     * @param jsonParams 调用的参数
-     * @return
-     * @throws Exception
-     */
-    public static CloseableHttpResponse doPostResponse(String url, String jsonParams) throws Exception {
-        CloseableHttpResponse response = null;
-        HttpPost httpPost = new HttpPost(url);
-
-        try {
-            StringEntity entity = new StringEntity(jsonParams, "UTF-8");
-            entity.setContentEncoding("UTF-8");
-            entity.setContentType("application/json");
-
-            httpPost.setEntity(entity);
-            httpPost.setHeader("content-type", "application/json");
-            //如果要设置 Basic Auth 的话
-//        httpPost.setHeader("Authorization", getHeader());
-            response = httpClient.execute(httpPost);
-        } finally {
-            if (response != null) {
-                EntityUtils.consume(response.getEntity());
-            }
-        }
-        return response;
-    }
-
-
-    public static String doPostString(String url, String jsonParams) throws Exception {
+    public static String doPutString(String url, String jsonParams) throws Exception {
         CloseableHttpResponse response = null;
-        HttpPost httpPost = new HttpPost(url);
+        HttpPut httpPut = new HttpPut(url);
 
         String httpStr;
         try {
             StringEntity entity = new StringEntity(jsonParams, "UTF-8");
             entity.setContentEncoding("UTF-8");
             entity.setContentType("application/json");
 
-            httpPost.setEntity(entity);
-            httpPost.setHeader("content-type", "application/json");
+            httpPut.setEntity(entity);
+            httpPut.setHeader("content-type", "application/json");
             //如果要设置 Basic Auth 的话
-//        httpPost.setHeader("Authorization", getHeader());
-            response = httpClient.execute(httpPost);
+//        httpPut.setHeader("Authorization", getHeader());
+            response = httpClient.execute(httpPut);
             httpStr = EntityUtils.toString(response.getEntity(), "UTF-8");
 
         } finally {
@@ -118,30 +114,34 @@ public static String doPostString(String url, String jsonParams) throws Exceptio
     }
 
 
-    public static String doPutString(String url, String jsonParams) throws Exception {
+    /**
+     * 发送 POST 请求（HTTP），JSON形式
+     *
+     * @param url        调用的地址
+     * @param jsonParams 调用的参数
+     * @return
+     * @throws Exception
+     */
+    public static CloseableHttpResponse doPostResponse(String url, String jsonParams) throws Exception {
         CloseableHttpResponse response = null;
-        HttpPut httpPut = new HttpPut(url);
+        HttpPost httpPost = new HttpPost(url);
 
-        String httpStr;
         try {
             StringEntity entity = new StringEntity(jsonParams, "UTF-8");
             entity.setContentEncoding("UTF-8");
             entity.setContentType("application/json");
 
-            httpPut.setEntity(entity);
-            httpPut.setHeader("content-type", "application/json");
+            httpPost.setEntity(entity);
+            httpPost.setHeader("content-type", "application/json");
             //如果要设置 Basic Auth 的话
-//        httpPut.setHeader("Authorization", getHeader());
-            response = httpClient.execute(httpPut);
-            httpStr = EntityUtils.toString(response.getEntity(), "UTF-8");
-
+//        httpPost.setHeader("Authorization", getHeader());
+            response = httpClient.execute(httpPost);
         } finally {
             if (response != null) {
                 EntityUtils.consume(response.getEntity());
-                response.close();
             }
         }
-        return httpStr;
+        return response;
     }
 
 
@@ -156,4 +156,4 @@ private static String getHeader() {
         String authHeader = "Basic " + new String(encodedAuth);
         return authHeader;
     }
-}
+}

flink-learning-common/src/main/java/com/zhisheng/common/utils/KafkaConfigUtil.java

@@ -100,8 +100,4 @@ private static Map<KafkaTopicPartition, Long> buildOffsetByTime(Properties props
         consumer.close();
         return partitionOffset;
     }
-
-    public static SingleOutputStreamOperator<MetricEvent> parseSource(DataStreamSource<MetricEvent> dataStreamSource) {
-        return dataStreamSource.assignTimestampsAndWatermarks(new MetricWatermark());
-    }
 }

flink-learning-connectors/flink-learning-connectors-akka/pom.xml

@@ -15,7 +15,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
     </dependencies>
 </project>

flink-learning-connectors/flink-learning-connectors-es2/src/main/java/com/zhisheng/connectors/es2/Sink2ES2Main.java

@@ -1,7 +1,7 @@
 package com.zhisheng.connectors.es2;
 
 /**
- * Desc:
+ * Desc: sink data to es2
  * Created by zhisheng on 2019/10/22 下午5:10
  * blog：http://www.54tianzhisheng.cn/
  * 微信公众号：zhisheng

flink-learning-connectors/flink-learning-connectors-es5/src/main/java/com/zhisheng/connectors/es5/Sink2ES5Main.java

@@ -1,7 +1,7 @@
 package com.zhisheng.connectors.es5;
 
 /**
- * Desc:
+ * Desc: sink data to es5
  * Created by zhisheng on 2019/10/22 下午5:10
  * blog：http://www.54tianzhisheng.cn/
  * 微信公众号：zhisheng

flink-learning-connectors/flink-learning-connectors-es6/src/main/java/com/zhisheng/connectors/es6/Sink2ES6Main.java

@@ -21,6 +21,7 @@
 import static com.zhisheng.common.constant.PropertiesConstants.*;
 
 /**
+ * sink data to es6
  * blog：http://www.54tianzhisheng.cn/
  * 微信公众号：zhisheng
  */

flink-learning-connectors/flink-learning-connectors-es6/src/main/java/com/zhisheng/connectors/es6/utils/ESSinkUtil.java

@@ -12,7 +12,7 @@
 import java.util.List;
 
 /**
- * Desc: ES Sink util（get ES host、addSink）//todo: index template & x-pack
+ * Desc: ES Sink utils（get ES host、addSink）//todo: index template & x-pack
  * Created by zhisheng on 2019/10/21 下午3:05
  * blog：http://www.54tianzhisheng.cn/
  * 微信公众号：zhisheng

flink-learning-connectors/flink-learning-connectors-es7/src/main/java/com/zhisheng/connectors/es7/Sink2ES7Main.java

@@ -1,7 +1,7 @@
 package com.zhisheng.connectors.es7;
 
 /**
- * Desc:
+ * Desc: sink data to es7
  * Created by zhisheng on 2019/10/22 下午5:10
  * blog：http://www.54tianzhisheng.cn/
  * 微信公众号：zhisheng

flink-learning-connectors/flink-learning-connectors-kudu/pom.xml

@@ -20,7 +20,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.kudu</groupId>

flink-learning-connectors/pom.xml

@@ -38,7 +38,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
     </dependencies>
 

flink-learning-data-sinks/pom.xml

@@ -15,7 +15,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>mysql</groupId>

flink-learning-data-sources/pom.xml

@@ -15,7 +15,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>mysql</groupId>

flink-learning-examples/pom.xml

@@ -15,7 +15,7 @@
         <dependency>
             <groupId>com.zhisheng.flink</groupId>
             <artifactId>flink-learning-common</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>mysql</groupId>

flink-learning-examples/src/main/resources/application.properties

@@ -1,14 +1,15 @@
 kafka.brokers=localhost:9092
 kafka.group.id=zhisheng
 kafka.zookeeper.connect=localhost:2181
-metrics.topic=zhisheng
+metrics.topic=zhisheng_metrics
 stream.parallelism=5
 stream.checkpoint.interval=1000
 stream.checkpoint.enable=false
 #stream.checkpoint.type=memory
-stream.checkpoint.type=fs
-#stream.checkpoint.type=rocksdb
-stream.checkpoint.dir=file:///usr/local/state/
+#stream.checkpoint.type=fs
+stream.checkpoint.type=rocksdb
+#stream.checkpoint.dir=file:///usr/local/state/
+stream.checkpoint.dir=/Users/zhisheng/Desktop
 
 
 #mysql

flink-learning-libraries/flink-learning-libraries-cep/pom.xml

@@ -12,7 +12,12 @@
     <artifactId>flink-learning-libraries-cep</artifactId>
 
     <dependencies>
-
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-cep_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <!--<scope>provided</scope>-->
+        </dependency>
     </dependencies>
 
     <build>
@@ -36,7 +41,7 @@
 
                                 <transformer
                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                                    <mainClass>com.zhisheng.libraries.cep.Main</mainClass>
+                                    <mainClass>com.zhisheng.libraries.cep.CEPMain</mainClass>
                                 </transformer>
                                 <transformer
                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">

flink-learning-libraries/flink-learning-libraries-cep/src/main/resources/application.properties

@@ -1,7 +1,6 @@
 kafka.brokers=localhost:9092
-kafka.group.id=metrics-group
+kafka.group.id=zhisheng
 kafka.zookeeper.connect=localhost:2181
-metrics.topic=alert-metrics
-stream.parallelism=5
-stream.checkpoint.interval=1000
+metrics.topic=zhisheng_metrics
+stream.parallelism=1
 stream.checkpoint.enable=false

flink-learning-libraries/flink-learning-libraries-machine-learning/README.md

@@ -2,4 +2,16 @@
 
 TODO：
 
-因在 Flink 1.9 版本中移除了 flink-ml 模块，所以暂不对这块做讲解，后面升级 Flink 版本时，该模块可以加入更多东西，doing！
+因在 Flink 1.9 版本中移除了 flink-ml 模块，所以暂不对这块做讲解，后面升级 Flink 版本时，该模块可以加入更多东西，doing！
+
+//可以根据 1.8 版本的 Flink-ml 进行学习，引入依赖：
+
+```xml
+<dependency>
+    <groupId>org.apache.flink</groupId>
+    <artifactId>flink-ml_${scala.binary.version}</artifactId>
+    <version>1.8.0</version>
+</dependency>
+```
+
+haberman.data 数据集是 '乳腺癌手术患者生存率的研究的病例'，数据以逗号分隔的，其中前三列是特征，最后一列是类别（患者存活 5 年或更长时间的是 1，在 5 年内死亡的是 2）。