update codes

src/main/java/com/gmei/data/ctr/CtrEstimateMainTagTest.java

@@ -31,7 +31,7 @@ public class CtrEstimateMainTagTest {
         String checkpointPath = parameterTool.get("checkpointPath","hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint");
         Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",true);
         Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
-        String startTime = parameterTool.get("startTime","2020-04-04 17:42:00");
+        String startTime = parameterTool.get("startTime","2020-04-04 20:42:00");
         Integer parallelism = parameterTool.getInt("parallelism",2);
 
         System.out.println("**********************************************************");

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

@@ -111,7 +111,7 @@ public class CtrEstimateTagOperator implements BaseOperator{
                         return false;
                     }
                 });
-        filter.print();
+        //filter.print();
         SingleOutputStreamOperator map = filter
                 .map(new MapFunction<JSONObject, CtrEstimateTagEtl>() {
                     @Override

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

@@ -600,6 +600,7 @@ public class CtrEstimateTagMysqlSink extends RichSinkFunction<DeviceCurrentEstim
                                         "last_update_time" +
                                         ") values ('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s')",
                                 deviceCurrentEstimateTag.getDeviceId(),
+                                commodityProjectObject.toJSONString(),
                                 commodityFirstDemandsObject.toJSONString(),
                                 commodityFirstPositionsObject.toJSONString(),
                                 commodityFirstSolutionsObject.toJSONString(),