update codes

bin/start_tag.sh

@@ -27,7 +27,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \
 --retryInteral 3000 \
 --checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint' \
 --parallelism 6 \
---startTime '2020-04-04 17:00:00' \
+--startTime '2020-04-04 17:42:00' \
 >> /data/log/ctr-estimate/ctr-estimate-tag.out 2>&1 &
 
 tail -f /data/log/ctr-estimate/ctr-estimate-tag.out

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

@@ -32,7 +32,7 @@ public class CtrEstimateMainTag {
         Integer maxRetry = parameterTool.getInt("maxRetry",3);
         Long retryInteral = parameterTool.getLong("retryInteral",3000);
         String checkpointPath = parameterTool.get("checkpointPath","hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate/checkpoint");
-        Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",false);
+        Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",true);
         Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
         String startTime = parameterTool.get("startTime");
         Integer parallelism = parameterTool.getInt("parallelism",2);

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

@@ -168,7 +168,6 @@ public class CtrEstimateTagOperator implements BaseOperator{
                 .uid("zhengxingAsyncDataStream")
                 .setParallelism(parallelism);
         DataStream<DeviceCurrentEstimateTagTmp> asyncDataStream = tidbAsyncDataStream.union(zhengxingAsyncDataStream);
-        //asyncDataStream.print();
 
         asyncDataStream
         .addSink(new CtrEstimateTagMysqlSink(outJdbcUrl, maxRetry, retryInteral))

src/main/java/com/gmei/data/ctr/source/TidbMysqlAsyncSource.java

@@ -89,6 +89,7 @@ public class TidbMysqlAsyncSource extends RichAsyncFunction<CtrEstimateTagEtl, D
             }
             if(StringUtils.isNotBlank(sql)){
                 dcett = findTagInfo(sql);
+                if(null != dcett){
                     Date date = new Date();
                     dcett.setType(type);
                     dcett.setDeviceId(ctrEstimateTagEtl.getDeviceId());
@@ -96,6 +97,7 @@ public class TidbMysqlAsyncSource extends RichAsyncFunction<CtrEstimateTagEtl, D
                     dcett.setLastUpdateTime(DateUtils.getTimeStr(date));
                 }
             }
+        }
         return dcett;
     }
 

src/main/java/com/gmei/data/ctr/source/ZhengxingMysqlAsyncSource.java

@@ -122,12 +122,14 @@ public class ZhengxingMysqlAsyncSource extends RichAsyncFunction<CtrEstimateTagE
         }
         if(StringUtils.isNotBlank(sql)){
             dcett = findTagInfo(sql,keyWord);
+            if(null != dcett){
                 Date date = new Date();
                 dcett.setType(type);
                 dcett.setDeviceId(ctrEstimateTagEtl.getDeviceId());
                 dcett.setPartitionDate(DateUtils.getDateStr(date));
                 dcett.setLastUpdateTime(DateUtils.getTimeStr(date));
             }
+        }
         return dcett;
     }