会员标签4.0

src/main/java/com/gic/spark/datasource/entity/DataSourceSharding.java

@@ -52,7 +52,7 @@ public class DataSourceSharding extends DataSourceEntity {
         SparkSession sparkSession = SparkEnvManager.getInstance().getSparkSession();
         ShardingRddManager shardingRddManager = new ShardingRddManager(shardingConfig,sparkSession);
         Dataset<Row> sourceDataset = shardingRddManager.getDatasetByEnterpriseIds(targetTable, enterpriseList, null, null).repartition(new Column("enterprise_id"));
-        SparkHiveUtil.createHivePartitionTable(sourceDataset, "enterprise_id", "tag", shardingConfig.get_id().replaceAll("\\.", "_") + "_" + targetTable);
+        SparkHiveUtil.createHivePartitionTable(sourceDataset, "enterprise_id", "tag", shardingConfig.get_id().replaceAll("\\.", "_") + "_" + targetTable,sparkSession);
 
         HivePartitionUtil.saveDatasetToPartitionTable(sparkSession, sourceDataset, getHiveTableName());
 

src/main/java/com/gic/spark/entity/bean/TagIntegralBean.java

+package com.gic.spark.entity.bean;
+
+import java.io.Serializable;
+import java.sql.Timestamp;
+
+/**
+ * @description:
+ * @author: wangxk
+ * @date: 2020/4/27
+ */
+public class TagIntegralBean implements Serializable {
+
+    private int enterprise_Id;
+    private int cu_Type;
+    private int cu_Id;
+    private int integral_Value;
+    private java.sql.Timestamp create_Time;
+
+    public int getEnterprise_Id() {
+        return enterprise_Id;
+    }
+
+    public void setEnterprise_Id(int enterprise_Id) {
+        this.enterprise_Id = enterprise_Id;
+    }
+
+    public int getCu_Type() {
+        return cu_Type;
+    }
+
+    public void setCu_Type(int cu_Type) {
+        this.cu_Type = cu_Type;
+    }
+
+    public int getCu_Id() {
+        return cu_Id;
+    }
+
+    public void setCu_Id(int cu_Id) {
+        this.cu_Id = cu_Id;
+    }
+
+    public int getIntegral_Value() {
+        return integral_Value;
+    }
+
+    public void setIntegral_Value(int integral_Value) {
+        this.integral_Value = integral_Value;
+    }
+
+    public Timestamp getCreate_Time() {
+        return create_Time;
+    }
+
+    public void setCreate_Time(Timestamp create_Time) {
+        this.create_Time = create_Time;
+    }
+}

src/main/java/com/gic/spark/entity/request/FilterRequest.java

@@ -3,6 +3,7 @@ package com.gic.spark.entity.request;
 import com.gic.spark.entity.enumtype.DomainType;
 
 import java.io.Serializable;
+import java.util.List;
 
 /**
  * Created by paste on 2018/7/25 11:26
@@ -13,6 +14,8 @@ public abstract class FilterRequest implements Serializable{
 
     protected DomainType domainType;
 
+    protected List<String> cuVals;
+
     public DomainType getDomainType() {
         return domainType;
     }
@@ -20,4 +23,12 @@ public abstract class FilterRequest implements Serializable{
     public void setDomainType(DomainType domainType) {
         this.domainType = domainType;
     }
+
+    public List<String> getCuVals() {
+        return cuVals;
+    }
+
+    public void setCuVals(List<String> cuVals) {
+        this.cuVals = cuVals;
+    }
 }

src/main/java/com/gic/spark/entity/request/IntegralRequest.java

@@ -15,16 +15,16 @@ public class IntegralRequest extends FilterRequest {
     /**
      * 最小数字
      */
-    private Double beginNum;
+    private Integer beginNum;
     /**
      * 最大数字
      */
-    private Double endNum;
+    private Integer endNum;
 
     /**
      * 相等数字
      */
-    private Double equalNum;
+    private Integer equalNum;
 
     private NumberType numberType;
 
@@ -43,29 +43,6 @@ public class IntegralRequest extends FilterRequest {
 
     private TimeRangeType timeRangeType;
 
-    public Double getBeginNum() {
-        return beginNum;
-    }
-
-    public void setBeginNum(Double beginNum) {
-        this.beginNum = beginNum;
-    }
-
-    public Double getEndNum() {
-        return endNum;
-    }
-
-    public void setEndNum(Double endNum) {
-        this.endNum = endNum;
-    }
-
-    public Double getEqualNum() {
-        return equalNum;
-    }
-
-    public void setEqualNum(Double equalNum) {
-        this.equalNum = equalNum;
-    }
 
     public NumberType getNumberType() {
         return numberType;
@@ -106,4 +83,28 @@ public class IntegralRequest extends FilterRequest {
     public void setTimeRangeType(TimeRangeType timeRangeType) {
         this.timeRangeType = timeRangeType;
     }
+
+    public Integer getBeginNum() {
+        return beginNum;
+    }
+
+    public void setBeginNum(Integer beginNum) {
+        this.beginNum = beginNum;
+    }
+
+    public Integer getEndNum() {
+        return endNum;
+    }
+
+    public void setEndNum(Integer endNum) {
+        this.endNum = endNum;
+    }
+
+    public Integer getEqualNum() {
+        return equalNum;
+    }
+
+    public void setEqualNum(Integer equalNum) {
+        this.equalNum = equalNum;
+    }
 }

src/main/java/com/gic/spark/filter/AbstractTagFilter.java

+package com.gic.spark.filter;
+
+import com.gic.spark.entity.request.FilterRequest;
+import com.gic.spark.tag.TagConstant;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.Column;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+
+/**
+ * @description:
+ * @author: wangxk
+ * @date: 2020/4/27
+ */
+public abstract class AbstractTagFilter implements BaseTagFilter{
+
+    protected static boolean cuHandle(FilterRequest request, int cuType){
+            boolean result=false;
+                switch (request.getDomainType()){
+                    case ACU_INFO:
+                            result=TagConstant.CU_TYPE_ACU==cuType;
+                        break;
+                    case SCU_INFO:
+                            result=TagConstant.CU_TYPE_SCU==cuType;
+                        break;
+                    case MCU_INFO:
+                            result=TagConstant.CU_TYPE_MCU==cuType;
+                        break;
+                    default:break;
+                }
+            return result;
+    }
+}

src/main/java/com/gic/spark/filter/TagAboutExpireIntegralFilter.java

 package com.gic.spark.filter;
 
 import com.gic.spark.datasource.entity.DataSourceEntity;
+import com.gic.spark.datasource.entity.DataSourceSharding;
 import com.gic.spark.entity.request.FilterRequest;
 import org.apache.spark.api.java.JavaRDD;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -13,6 +15,7 @@ import java.util.List;
  */
 public class TagAboutExpireIntegralFilter implements BaseTagFilter{
 
+    private DataSourceSharding memberSharding=new DataSourceSharding("member_sharding_4.0_test","tab_integral_cu_change_log");
     private static TagAboutExpireIntegralFilter instance;
 
     public static TagAboutExpireIntegralFilter getInstance() {
@@ -25,11 +28,15 @@ public class TagAboutExpireIntegralFilter implements BaseTagFilter{
     private TagAboutExpireIntegralFilter(){}
     @Override
     public List<DataSourceEntity> necessarySourceList() {
-        return null;
+        List<DataSourceEntity> result = new ArrayList();
+        result.add(memberSharding);
+        return result;
     }
 
     @Override
     public JavaRDD<Integer> filterValidMember(Integer enterpriseId, FilterRequest filterOption) {
+
+
         return null;
     }
 }

src/main/java/com/gic/spark/tag/TagConstant.java

@@ -8,7 +8,6 @@ package com.gic.spark.tag;
 public class TagConstant {
 
 
-    public static final String TAG_COMPUTE_TYPE_BETWEEN = "between";//区间筛选
     public static final String TAG_COMPUTE_TYPE_EQUAL = "=";
     public static final String TAG_COMPUTE_TYPE_LESS_EQUAL = "<=";
     public static final String TAG_COMPUTE_TYPE_LARGER_EQUAL = ">=";
@@ -89,6 +88,18 @@ public class TagConstant {
     public static final String TAG_KEY_TYPE_TIME_RANGE="timeRange";
     public static final String TAG_KEY_TYPE_NUMBER="number";
 
+    public static final String TAG_TIME_TYPE_FIXATION="fixation";
+    public static final String TAG_TIME_TYPE_LATELY="lately";
+    public static final String TAG_TIME_TYPE_LATER="later";
 
+    public static final String TAG_COMPUTE_TYPE_GT="gt"; //大于
+    public static final String TAG_COMPUTE_TYPE_GTE="gte"; //大于等于
+    public static final String TAG_COMPUTE_TYPE_LT="lt"; //小于
+    public static final String TAG_COMPUTE_TYPE_LTE="lte"; //小于等于
+    public static final String TAG_COMPUTE_TYPE_EQ="eq"; //等于
+    public static final String TAG_COMPUTE_TYPE_BETWEEN="between"; //范围
 
+    public static final int CU_TYPE_ACU=1;
+    public static final int CU_TYPE_SCU=2;
+    public static final int CU_TYPE_MCU=3;
 }

src/main/java/com/gic/spark/tag/TagValueParser.java

@@ -3,11 +3,15 @@ package com.gic.spark.tag;
 import com.gic.spark.entity.TagConditionDTO;
 import com.gic.spark.entity.TagConditionValDTO;
 import com.gic.spark.entity.enumtype.DomainType;
+import com.gic.spark.entity.enumtype.NumberType;
 import com.gic.spark.entity.enumtype.TimeRangeType;
 import com.gic.spark.entity.request.FilterRequest;
 import com.gic.spark.entity.request.IntegralRequest;
-import sun.util.resources.es.CalendarData_es;
+import com.gic.spark.util.DateUtil;
+import com.google.common.collect.Lists;
 
+import java.sql.Timestamp;
+import java.util.Arrays;
 import java.util.List;
 
 /**
@@ -24,7 +28,7 @@ public class TagValueParser {
                 request=getIntegralRequest(conditionDTO.getTagTemplateElInfo());
                 break;
             case TagConstant.TAG_CODE_ABOUT_EXPIRE_INTEGRAL:
-
+                request=getIntegralRequest(conditionDTO.getTagTemplateElInfo());
                 break;
             default:break;
         }
@@ -36,27 +40,66 @@ public class TagValueParser {
         for(TagConditionValDTO conditionValDTO:conditionValDTOList){
             String[] keys=conditionValDTO.getKey().split(".");
             if(keys.length==1){
-                setDomainHandle(request,keys[0]);
+                setDomainHandle(request,keys[0],conditionValDTO.getVal());
             }else{
                 switch (keys[0]){
                     case TagConstant.TAG_KEY_TYPE_TIME_RANGE:
                         switch (keys[1]){
-
+                            case TagConstant.TAG_TIME_TYPE_FIXATION:
+                                String[] vals=conditionValDTO.getVal().split(",");
+                                request.setBeginTime(new Timestamp(DateUtil.stringToDate(vals[0]).getTime()));
+                                request.setEndTime(new Timestamp(DateUtil.stringToDate(vals[1]).getTime()));
+                                request.setTimeRangeType(TimeRangeType.FIXATION);
+                                break;
+                            case TagConstant.TAG_TIME_TYPE_LATELY:
+                                request.setTimeNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setTimeRangeType(TimeRangeType.LATELY);
+                                break;
+                            case TagConstant.TAG_TIME_TYPE_LATER:
+                                request.setTimeNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setTimeRangeType(TimeRangeType.LATER);
+                                break;
+                            default:break;
                         }
                         break;
                     case  TagConstant.TAG_KEY_TYPE_NUMBER:
-
-                        break;
-
+                        switch (keys[1]){
+                            case TagConstant.TAG_COMPUTE_TYPE_GT:
+                                request.setBeginNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setNumberType(NumberType.gt);
+                                break;
+                            case TagConstant.TAG_COMPUTE_TYPE_GTE:
+                                request.setBeginNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setNumberType(NumberType.gte);
+                                break;
+                            case TagConstant.TAG_COMPUTE_TYPE_LT:
+                                request.setEndNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setNumberType(NumberType.lt);
+                                break;
+                            case TagConstant.TAG_COMPUTE_TYPE_LTE:
+                                request.setEndNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setNumberType(NumberType.lte);
+                                break;
+                            case TagConstant.TAG_COMPUTE_TYPE_EQ:
+                                request.setEqualNum(Integer.parseInt(conditionValDTO.getVal()));
+                                request.setNumberType(NumberType.eq);
+                                break;
+                            case TagConstant.TAG_COMPUTE_TYPE_BETWEEN:
+                                String[] vals=conditionValDTO.getVal().split(",");
+                                request.setBeginNum(Integer.parseInt(vals[0]));
+                                request.setEndNum(Integer.parseInt(vals[1]));
+                                request.setNumberType(NumberType.between);
+                                break;
+                            default:break;
+                        }
                     default:break;
                 }
             }
-
         }
         return request;
     }
 
-    private static void setDomainHandle(FilterRequest request,String key){
+    private static void setDomainHandle(FilterRequest request,String key,String val){
         switch (key){
             case TagConstant.TAG_DOMAIN_ACU_INFO:
                 request.setDomainType(DomainType.ACU_INFO);
@@ -69,5 +112,7 @@ public class TagValueParser {
                 break;
             default:break;
         }
+        String[] vals=val.split(",");
+        request.setCuVals(Arrays.asList(vals));
     }
 }

src/main/java/com/gic/spark/util/DateUtil.java

@@ -1369,6 +1369,14 @@ public class DateUtil {
         return tmpString;
     }
 
+    public static Date getDate(){
+        Calendar calendar=Calendar.getInstance();
+        calendar.set(Calendar.HOUR_OF_DAY,0);
+        calendar.set(Calendar.MINUTE,0);
+        calendar.set(Calendar.SECOND,0);
+        return calendar.getTime();
+    }
+
     public static void main(String[] args) {
        /* Date date = setTimeHHmmss(new Date(), "08:00:00");
         System.out.println(datetimeToString(date));