update codes

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

@@ -158,6 +158,6 @@ public class CtrPfrCrtOperator implements BaseOperator{
 
         tidbAsyncDataStream
         .addSink(new CtrPfrCrtMysqlSink(outJdbcUrl, maxRetry, retryInteral))
-        .setParallelism(parallelism);
+        .setParallelism(1);
     }
 }

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

@@ -96,7 +96,7 @@ public class CtrPfrCrtMysqlSink extends RichSinkFunction<CtrPfrCrtBean> {
                                 ctrPfrCrtBean.getDeviceId(),
                                 ctrPfrCrtBean.getStatisticsType(),
                                 ctrPfrCrtBean.getStatisticsTypeId(),
-                                ctrPfrCrtBean.getPartitionDate()
+                                DateUtils.getDateStr(date)
                         )
                 );
                 if (resultSet.next()){
@@ -120,7 +120,8 @@ public class CtrPfrCrtMysqlSink extends RichSinkFunction<CtrPfrCrtBean> {
                                             "second_demands_pfr = '%s'," +
                                             "second_positions_pfr = '%s'," +
                                             "second_solutions_pfr = '%s'," +
-                                            "last_update_time = '%s'",
+                                            "last_update_time = '%s' " +
+                                            "where device_id = '%s' and statistics_type = '%s' and statistics_type_id = '%s' and partition_date = '%s'",
                                     StringUtils.increasePfr(projectPfr,ctrPfrCrtBean.getProjectPfr()),
                                     StringUtils.increasePfr(firstDemandsPfr,ctrPfrCrtBean.getFirstDemandsPfr()),
                                     StringUtils.increasePfr(firstPositionsPfr,ctrPfrCrtBean.getFirstPositionsPfr()),
@@ -128,7 +129,11 @@ public class CtrPfrCrtMysqlSink extends RichSinkFunction<CtrPfrCrtBean> {
                                     StringUtils.increasePfr(secondDemandsPfr,ctrPfrCrtBean.getSecondDemandsPfr()),
                                     StringUtils.increasePfr(secondPositionsPfr,ctrPfrCrtBean.getSecondPositionsPfr()),
                                     StringUtils.increasePfr(secondSolutionsPfr,ctrPfrCrtBean.getSecondSolutionsPfr()),
-                                    DateUtils.getTimeStr(date)
+                                    DateUtils.getTimeStr(date),
+                                    ctrPfrCrtBean.getDeviceId(),
+                                    ctrPfrCrtBean.getStatisticsType(),
+                                    ctrPfrCrtBean.getStatisticsTypeId(),
+                                    DateUtils.getDateStr(date)
                             )
                     );
                 }else{

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

@@ -3,8 +3,6 @@ package com.gmei.data.ctr.source;
 import com.alibaba.druid.pool.DruidDataSource;
 import com.gmei.data.ctr.bean.CtrPfrCrtBean;
 import com.gmei.data.ctr.bean.CtrPfrCrtEtlBean;
-import com.gmei.data.ctr.bean.CtrPfrRctEtlBean;
-import com.gmei.data.ctr.bean.CtrPfrRctBean;
 import com.gmei.data.ctr.common.Constants;
 import com.gmei.data.ctr.utils.DateUtils;
 import org.apache.commons.lang3.StringUtils;