update codes

src/main/java/com/gmei/data/ctr/operator/CtrEstimatePfrOperator.java

@@ -49,22 +49,23 @@ public class CtrEstimatePfrOperator implements BaseOperator{
     }
     @Override
     public void run() {
-        SingleOutputStreamOperator jsonStream = dataStream
+        SingleOutputStreamOperator filter01 = dataStream
                 .filter(new FilterFunction<String>() {
                     @Override
                     public boolean filter(String value) throws Exception {
                         return JSON.isValid(value);
                     }
-                })
+                }).setParallelism(parallelism);
+        SingleOutputStreamOperator map01 = filter01
                 .map(new MapFunction<String, JSONObject>() {
                     @Override
                     public JSONObject map(String value) throws Exception {
                         return JSON.parseObject(value);
                     }
-                });
+                }).setParallelism(parallelism);
     // jsonStream.print();
-    SingleOutputStreamOperator filter =
-        jsonStream.filter(
+    SingleOutputStreamOperator filter2 =
+            map01.filter(
             new FilterFunction<JSONObject>() {
               @Override
               public boolean filter(JSONObject jsonObject) throws Exception {
@@ -113,7 +114,7 @@ public class CtrEstimatePfrOperator implements BaseOperator{
               }
             }).setParallelism(parallelism);
         //filter.print();
-        SingleOutputStreamOperator map = filter
+        SingleOutputStreamOperator map02 = filter2
                 .map(new MapFunction<JSONObject, CtrEstimatePfrEtl>() {
                     @Override
                     public CtrEstimatePfrEtl map(JSONObject jsonObject) throws Exception {
@@ -158,7 +159,7 @@ public class CtrEstimatePfrOperator implements BaseOperator{
                 }).setParallelism(parallelism);
         //map.print();
         DataStream<DeviceCurrentEstimatePfrTmp> tidbAsyncDataStream = AsyncDataStream
-                .unorderedWait(map, new TidbMysqlAsyncPfrSource(jerryJdbcUrl,jerryUsername,jerryPassword), 1, TimeUnit.MINUTES, 1000)
+                .unorderedWait(map02, new TidbMysqlAsyncPfrSource(jerryJdbcUrl,jerryUsername,jerryPassword), 1, TimeUnit.MINUTES, 1000)
                 .uid("tidbAsyncDataStream")
                 .setParallelism(parallelism);
 

src/main/java/com/gmei/data/ctr/sink/CtrEstimatePfrMysqlSink.java

@@ -64,21 +64,12 @@ public class CtrEstimatePfrMysqlSink extends RichSinkFunction<DeviceCurrentEstim
         super.close();
     }
 
-    /**
-     * 最近偏好持久化
-     * @param deviceCurrentEstimatePfrTmp
-     */
-    private void insertAndDel(DeviceCurrentEstimatePfrTmp deviceCurrentEstimatePfrTmp)  {
-        insert(deviceCurrentEstimatePfrTmp);
-        del(deviceCurrentEstimatePfrTmp);
-    }
-
     /**
      * 插入最新数据
      * @param deviceCurrentEstimatePfrTmp
      * @throws SQLException
      */
-    private void insert(DeviceCurrentEstimatePfrTmp deviceCurrentEstimatePfrTmp)  {
+    private void insertAndDel(DeviceCurrentEstimatePfrTmp deviceCurrentEstimatePfrTmp)  {
         Statement statement = null;
         Date date = new Date();
         try{
@@ -113,31 +104,6 @@ public class CtrEstimatePfrMysqlSink extends RichSinkFunction<DeviceCurrentEstim
                             DateUtils.getTimeStr(date)
                     )
             );
-        }catch (Exception e){
-            e.printStackTrace();
-        }finally {
-            try{
-                if (statement != null) {
-                    statement.close();
-                }
-                if (connection != null) {
-                    connection.close();
-                }
-            }catch (Exception e){
-                e.printStackTrace();
-            }
-        }
-    }
-
-    /**
-     * 删除过期数据
-     * @param deviceCurrentEstimatePfrTmp
-     */
-    private void del(DeviceCurrentEstimatePfrTmp deviceCurrentEstimatePfrTmp)  {
-        Statement statement = null;
-        Date date = new Date();
-        try{
-            statement = connection.createStatement();
             statement.executeUpdate(
                     String.format(
                             "delete from device_recently_estimate_view_pfr where " +