会员标签4.0

src/main/java/com/gic/spark/entity/table/TabSceneCrowd.java

@@ -3,16 +3,16 @@ package com.gic.spark.entity.table;
 
 public class TabSceneCrowd {
 
-  private int id;
+  private long id;
   private String scene_Crowd_Name;
   private int real_Time;
   private java.sql.Timestamp scene_Crowd_Expire_Time;
-  private int scene_Crowd_Category_Id;
+  private long scene_Crowd_Category_Id;
   private String scene_Crowd_Desc;
   private int scene_Crowd_Update_Type;
   private int scene_Crowd_Update_Type_Val;
   private java.sql.Timestamp update_Time;
-  private int member_Size;
+  private long member_Size;
   private java.sql.Timestamp create_Time;
   private int valid_Flag;
   private String tag_Condition_Group_Info;
@@ -21,11 +21,11 @@ public class TabSceneCrowd {
   private String crowd_Param;
 
 
-  public int getId() {
+  public long getId() {
     return id;
   }
 
-  public void setId(int id) {
+  public void setId(long id) {
     this.id = id;
   }
 
@@ -57,11 +57,11 @@ public class TabSceneCrowd {
   }
 
 
-  public int getScene_Crowd_Category_Id() {
+  public long getScene_Crowd_Category_Id() {
     return scene_Crowd_Category_Id;
   }
 
-  public void setScene_Crowd_Category_Id(int scene_Crowd_Category_Id) {
+  public void setScene_Crowd_Category_Id(long scene_Crowd_Category_Id) {
     this.scene_Crowd_Category_Id = scene_Crowd_Category_Id;
   }
 
@@ -102,11 +102,11 @@ public class TabSceneCrowd {
   }
 
 
-  public int getMember_Size() {
+  public long getMember_Size() {
     return member_Size;
   }
 
-  public void setMember_Size(int member_Size) {
+  public void setMember_Size(long member_Size) {
     this.member_Size = member_Size;
   }
 

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

@@ -2,7 +2,6 @@ package com.gic.spark.filter;
 
 import com.gic.spark.datasource.entity.DataSourceHive;
 import com.gic.spark.entity.bean.TagConsumeBean;
-import com.gic.spark.entity.request.AbstractFilterRequestInteger;
 import com.gic.spark.entity.request.TagConsumeRequest;
 import com.gic.spark.util.ConstantUtil;
 import org.apache.spark.api.java.JavaRDD;
@@ -15,7 +14,7 @@ import java.util.List;
  * @author: wangxk
  * @date: 2020/5/7
  */
-public abstract class AbstractTagConsumFilter implements BaseTagFilter{
+public abstract class AbstractTagConsumFilter implements TagFilter {
     DataSourceHive dataSourceHive = new DataSourceHive(ConstantUtil.ADS_GIC_TRD_ECU_SALES_LABEL_D);
 
     protected static JavaRDD<TagConsumeBean> statisticsTypeHandle(JavaRDD<TagConsumeBean> consumeRDD,  TagConsumeRequest consumeRequest){

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

@@ -15,7 +15,7 @@ import java.util.List;
  * @author: wangxk
  * @date: 2020/4/24
  */
-public class TagAboutExpireIntegralFilter implements BaseTagFilter{
+public class TagAboutExpireIntegralFilter implements TagFilter {
 
     private DataSourceSharding dataSourceSharding=new DataSourceSharding(AppEnvUtil.MEMBER_SHARDING_4, ConstantUtil.TAB_INTEGRAL_CU_CHANGE_LOG);
     private static TagAboutExpireIntegralFilter instance;

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

@@ -26,7 +26,7 @@ import java.util.Set;
  * @author: wangxk
  * @date: 2020/4/24
  */
-public class TagAccumulatedIntegralFilter implements BaseTagFilter{
+public class TagAccumulatedIntegralFilter implements TagFilter {
 
     private DataSourceSharding dataSourceSharding=new DataSourceSharding(AppEnvUtil.MEMBER_SHARDING_4, ConstantUtil.TAB_INTEGRAL_CU_CHANGE_LOG);
     private static TagAccumulatedIntegralFilter instance;

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

@@ -20,7 +20,7 @@ import java.util.Set;
  * @author: wangxk
  * @date: 2020/4/29
  */
-public class TagCouponFilter implements BaseTagFilter{
+public class TagCouponFilter implements TagFilter {
 
     private DataSourceSharding dataSourceSharding=new DataSourceSharding(AppEnvUtil.MARKETING_SHARDING, ConstantUtil.TAB_COUPON_LOG);
     private static TagCouponFilter instance;

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

@@ -22,7 +22,7 @@ import java.util.Set;
  * @author: wangxk
  * @date: 2020/4/20
  */
-public class TagCurrentCouponNumFilter implements BaseTagFilter{
+public class TagCurrentCouponNumFilter implements TagFilter {
 
     private DataSourceSharding dataSourceSharding=new DataSourceSharding(AppEnvUtil.MARKETING_SHARDING, ConstantUtil.TAB_COUPON_LOG);
     private static TagCurrentCouponNumFilter instance;

src/main/java/com/gic/spark/filter/BaseTagFilter.java → src/main/java/com/gic/spark/filter/TagFilter.java

@@ -11,7 +11,7 @@ import java.util.List;
  * <p>
  * TODO: class description
  */
-public interface BaseTagFilter {
+public interface TagFilter {
 
     List<DataSourceEntity> necessarySourceList();
 

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

+package com.gic.spark.tag;
+
+import com.gic.spark.filter.*;
+import com.gic.spark.util.DingtalkMessageUtil;
+
+/**
+ * @description:
+ * @author: wangxk
+ * @date: 2020/5/13
+ */
+public class TagFilterFactory {
+
+    private static TagFilterFactory instance;
+
+    public static TagFilterFactory getInstance() {
+        if(null==instance){
+            instance=new TagFilterFactory();
+        }
+        return instance;
+    }
+    private TagFilterFactory(){}
+
+    public TagFilter getTagFilter(String tagEsFieldName){
+        TagFilter tagFilter=null;
+        switch (tagEsFieldName) {
+            //积分信息
+            case TagConstant.TAG_CODE_ACCUMULATED_INTEGRAL:
+                tagFilter= TagAccumulatedIntegralFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_ABOUT_EXPIRE_INTEGRAL:
+                tagFilter= TagAboutExpireIntegralFilter.getInstance();
+                break;
+
+
+            //卡卷类
+            case TagConstant.TAG_CODE_CURRENT_COUPON_NUM:
+                tagFilter= TagCurrentCouponNumFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_UNCLAIMED_COUPON:
+                tagFilter= TagCouponFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_GET_NO_CANCEL_COUPON:
+                tagFilter= TagCouponFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_NO_CANCEL_EXPIRES_COUPON:
+                tagFilter= TagCouponFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_CANCEL_COUPON:
+                tagFilter=TagCouponFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_DONATION_IN_COUPON:
+                tagFilter=TagCouponFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_YET_DONATION_COUPON:
+                tagFilter=TagCouponFilter.getInstance();
+                break;
+
+
+            //消费统计
+            case TagConstant.TAG_CODE_CONSUMPTION_TIME:
+                tagFilter=TagConsumptionTimeFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_PER_CUSTOMER_TRANSACTION:
+                tagFilter=TagPerCustomerTransactionFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_ASSOCIATED_PURCHASE_RATE:
+                tagFilter=TagAssociatedPurchaseRateFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_AVERAGE_DISCOUNT_FACTOR:
+                tagFilter=TagAverageDiscountFactorFilter.getInstance();
+                break;
+            case TagConstant.TAG_CODE_CONSUMPTION_SLEEP_DAYS:
+                tagFilter=TagConsumptionSleepDaysFilter.getInstance();
+                break;
+            default:
+                DingtalkMessageUtil.sendAlertMessage("暂未支持的非实时标签: " + tagEsFieldName);
+                break;
+        }
+        return tagFilter;
+    }
+}

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

@@ -40,9 +40,6 @@ import scala.collection.JavaConversions;
 import java.util.*;
 import org.apache.spark.api.java.Optional;
 
-import javax.sound.midi.MidiSystem;
-import javax.swing.text.TabSet;
-
 import static com.gic.spark.datasource.mysql.MysqlDatasource.JDBC_OPTIONS;
 
 /**
@@ -114,7 +111,7 @@ public class TagProcess {
     public void process(boolean extractData){
 
         Map<Integer, List<TagProcessEntity>> tagGroupByEnterpriseMap = new HashMap<>();
-        Map<Long, BaseTagFilter> tagIdToFilterMap = new HashMap();
+        Map<Long, TagFilter> tagIdToFilterMap = new HashMap();
         for(SceneCrowdDTO sceneCrowdDTO:sceneCrowdDTOList){
             if(sceneCrowdDTO.getReal_Time()==1){
                 LinkedList<TagConditionGroupDTO> conditionGroupDTOS=sceneCrowdDTO.getConditionGroupDTOList();
@@ -127,7 +124,7 @@ public class TagProcess {
 
                     entity.tagList=conditionGroupDTOS.get(i).getConditionInfos();
                     for(TagConditionDTO conditionDTO:entity.tagList){//将tag同filter进行映射
-                        BaseTagFilter tagFilter=getTagFilter(conditionDTO.getTagEsFieldName());
+                        TagFilter tagFilter=TagFilterFactory.getInstance().getTagFilter(conditionDTO.getTagEsFieldName());
                         if(null!=tagFilter){
                             tagIdToFilterMap.put(conditionDTO.getTagId(),tagFilter);
                         }
@@ -155,6 +152,7 @@ public class TagProcess {
         }
 
         if (extractData) {
+            DataSourceManager.getInstance().init(isProduction);
             DataSourceManager.getInstance().extractDataToDatabase();
             DingtalkMessageUtil.sendAlertMessage("extractData to hive finish !");
         }
@@ -166,7 +164,7 @@ public class TagProcess {
             for(TagProcessEntity entity:enterpriseTagEntry.getValue()){
                 for(TagConditionDTO conditionDTO:entity.tagList){
                     if(tagIdToFilterMap.containsKey(conditionDTO.getTagId())){
-                        BaseTagFilter tagFilter=tagIdToFilterMap.get(conditionDTO.getTagId());
+                        TagFilter tagFilter=tagIdToFilterMap.get(conditionDTO.getTagId());
                         AbstractFilterRequest filterRequest=TagValueParser.parseFilterValue(conditionDTO);
 
                         final String groupId = entity.tagGroupId + "_" + conditionDTO.getTagId() + "_" + entity.level;
@@ -266,62 +264,4 @@ public class TagProcess {
         });
     }
 
-    private static BaseTagFilter getTagFilter(String tagEsFieldName){
-        BaseTagFilter tagFilter=null;
-        switch (tagEsFieldName) {
-            //积分信息
-            case TagConstant.TAG_CODE_ACCUMULATED_INTEGRAL:
-                tagFilter= TagAccumulatedIntegralFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_ABOUT_EXPIRE_INTEGRAL:
-                tagFilter= TagAboutExpireIntegralFilter.getInstance();
-                break;
-
-
-            //卡卷类
-            case TagConstant.TAG_CODE_CURRENT_COUPON_NUM:
-                tagFilter= TagCurrentCouponNumFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_UNCLAIMED_COUPON:
-                tagFilter= TagCouponFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_GET_NO_CANCEL_COUPON:
-                tagFilter= TagCouponFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_NO_CANCEL_EXPIRES_COUPON:
-                tagFilter= TagCouponFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_CANCEL_COUPON:
-                tagFilter=TagCouponFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_DONATION_IN_COUPON:
-                tagFilter=TagCouponFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_YET_DONATION_COUPON:
-                tagFilter=TagCouponFilter.getInstance();
-                break;
-
-
-            //消费统计
-            case TagConstant.TAG_CODE_CONSUMPTION_TIME:
-                tagFilter=TagConsumptionTimeFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_PER_CUSTOMER_TRANSACTION:
-                tagFilter=TagPerCustomerTransactionFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_ASSOCIATED_PURCHASE_RATE:
-                tagFilter=TagAssociatedPurchaseRateFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_AVERAGE_DISCOUNT_FACTOR:
-                tagFilter=TagAverageDiscountFactorFilter.getInstance();
-                break;
-            case TagConstant.TAG_CODE_CONSUMPTION_SLEEP_DAYS:
-                tagFilter=TagConsumptionSleepDaysFilter.getInstance();
-                break;
-            default:
-                DingtalkMessageUtil.sendAlertMessage("暂未支持的非实时标签: " + tagEsFieldName);
-                break;
-        }
-        return tagFilter;
-    }
 }

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

@@ -11,7 +11,7 @@ import java.util.List;
  */
 public class TagProcessEntity {
     int enterpriseId;
-    int tagGroupId;
+    long tagGroupId;
     int level;
     List<TagConditionDTO> tagList;
 }