会员标签4.0

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

@@ -32,6 +32,10 @@ public abstract class DataSourceEntity {
         return SparkEnvManager.getInstance().getSparkSession().sql(String.format("select * from %s where enterprise_id='%s'", getHiveTableName(), enterpriseId));
     }
 
+    public Dataset<Row> getDatasetByEntId(Integer enterpriseId) {
+        return SparkEnvManager.getInstance().getSparkSession().sql(String.format("select * from %s where ent_id='%s'", getHiveTableName(), enterpriseId));
+    }
+
     public abstract void extractDataToPartitionedHiveTable(List<Integer> enterpriseList);
 
 }

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

@@ -47,16 +47,14 @@ public class DataSourceHive extends DataSourceEntity {
     @Override
     public void extractDataToPartitionedHiveTable(List<Integer> enterpriseList) {
         SparkSession sparkSession = SparkEnvManager.getInstance().getSparkSession();
-        Dataset<Row> sourceDataset = sparkSession
-                .sql(String.format("select * from %s where enterprise_id in (%s)", sourceTable, Joiner.on("','").join(enterpriseList)))
-                .repartition(new Column("enterprise_id"));
+//        Dataset<Row> sourceDataset = sparkSession
+//                .sql(String.format("select * from %s where ent_id in (%s)", sourceTable, Joiner.on("','").join(enterpriseList)))
+//                .repartition(new Column("ent_id"));
 
-        if (table.equals("ads_gic_trd_ecu_sales_label_d")) {
-            sourceDataset = sparkSession
-                    .sql(String.format("select * from %s where enterprise_id in (%s) distribute by enterprise_id ", sourceTable, Joiner.on("','").join(enterpriseList)));
-        }
+        Dataset<Row> sourceDataset = sparkSession
+                    .sql(String.format("select * from %s where ent_id in (%s) distribute by ent_id ", sourceTable, Joiner.on("','").join(enterpriseList)));
 
-        SparkHiveUtil.createHivePartitionTable(sourceDataset, "enterprise_id", getSchema(), schema.replaceAll("\\.", "_") + "_" + table,sparkSession);
+        SparkHiveUtil.createHivePartitionTable(sourceDataset, "ent_id", getSchema(), schema.replaceAll("\\.", "_") + "_" + table,sparkSession);
 
         HivePartitionUtil.saveDatasetToPartitionTable(sparkSession, sourceDataset, getHiveTableName());
     }

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

@@ -14,7 +14,7 @@ import java.util.List;
  * @author: wangxk
  * @date: 2020/5/7
  */
-public abstract class AbstractTagConsumFilter implements TagFilter {
+public abstract class AbstractTagConsumFilter implements BaseTagFilter {
     DataSourceHive dataSourceHive = new DataSourceHive(ConstantUtil.ADS_GIC_TRD_ECU_SALES_LABEL_D);
 
     protected static JavaRDD<TrdEcuSalesLabelBean> statisticsTypeHandle(JavaRDD<TrdEcuSalesLabelBean> consumeRDD, TagConsumeRequest consumeRequest){

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

@@ -19,7 +19,7 @@ import java.util.List;
  * @author: wangxk
  * @date: 2020/8/3
  */
-public abstract class AbstractTagConsumRecordFilter implements TagFilter{
+public abstract class AbstractTagConsumRecordFilter implements BaseTagFilter {
 
      DataSourceHive dataSourceHive = new DataSourceHive(ConstantUtil.DWD_GIC_TRD_VIRTUAL_WDORDER_D);
 

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

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

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 TagFilter {
+public class TagAboutExpireIntegralFilter implements BaseTagFilter {
 
     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 TagFilter {
+public class TagAccumulatedIntegralFilter implements BaseTagFilter {
 
     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/TagAssociatedPurchaseRateFilter.java

@@ -37,7 +37,7 @@ public class TagAssociatedPurchaseRateFilter extends AbstractTagConsumFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeRequest consumeRequest=(TagConsumeRequest)request;
-        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
+        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
         consumeRDD=statisticsTypeHandle(consumeRDD,consumeRequest);
         JavaRDD<Long>ecuRdd=consumeRDD.mapToPair(data-> Tuple2.apply(data.getEcu_id(),data)).groupByKey()
                                         .mapPartitions(data->{

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

@@ -39,7 +39,7 @@ public class TagAverageDiscountFactorFilter extends AbstractTagConsumFilter {
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeRequest consumeRequest=(TagConsumeRequest)request;
-        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
+        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
         consumeRDD=statisticsTypeHandle(consumeRDD,consumeRequest);
         JavaRDD<Long>ecuRdd=consumeRDD.mapToPair(data-> Tuple2.apply(data.getEcu_id(),data)).groupByKey()
                                         .mapPartitions(data->{

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

@@ -46,14 +46,14 @@ public class TagConsumeCommodityFilter extends AbstractTagConsumRecordFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeCommodityRequest commodityRequest=(TagConsumeCommodityRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,commodityRequest);
         JavaPairRDD<Long,Long>orderRdd= consumeRecordRDD.filter(data-> StringUtils.isNotEmpty(data.getReceipts_time()))
                                                             .filter(data->checkTime(commodityRequest,DateUtil.strToDate(data.getReceipts_time(),DateUtil.FORMAT_DATETIME_19).getTime()))
                                                             .mapToPair(data-> Tuple2.apply(data.getVirtual_id(),data.getEcu_id()));
 
-        JavaPairRDD<Long,Long> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD()
+        JavaPairRDD<Long,Long> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEntId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD()
                                                                 .filter(data->{
                                                                     if(StringUtils.isNotEmpty(data.getSku_code())
                                                                         &&commodityRequest.getSkuCodeList().contains(data.getSku_code())){

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

@@ -42,7 +42,7 @@ public class TagConsumeTimeFilter extends AbstractTagConsumRecordFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeTimeRequest consumeTimeRequest=(TagConsumeTimeRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId)
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId)
                                                         .filter(new Column("is_eff_order").equalTo(1)), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeTimeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data-> StringUtils.isNotEmpty(data.getReceipts_time()))

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

@@ -41,7 +41,7 @@ public class TagConsumeTotalFilter extends AbstractTagConsumRecordFilter {
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD=MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD=MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
 
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);

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

@@ -38,7 +38,7 @@ public class TagConsumptionSleepDaysFilter extends AbstractTagConsumFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeRequest consumeRequest=(TagConsumeRequest)request;
-        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
+        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
         consumeRDD=statisticsTypeHandle(consumeRDD,consumeRequest);
         JavaRDD<Long>ecuRdd=consumeRDD.mapToPair(data-> Tuple2.apply(data.getEcu_id(),data.getSleep_days()))
                                         .reduceByKey((x,y)->x>y?y:x)

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

@@ -36,7 +36,7 @@ public class TagConsumptionTimeFilter extends AbstractTagConsumFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeRequest consumeRequest=(TagConsumeRequest)request;
-        JavaRDD<TrdEcuSalesLabelBean> consumeRDD=MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
+        JavaRDD<TrdEcuSalesLabelBean> consumeRDD=MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
         consumeRDD=statisticsTypeHandle(consumeRDD,consumeRequest);
         JavaRDD<Long>ecuRdd=consumeRDD.mapToPair(data-> Tuple2.apply(data.getEcu_id(),data.getOrder_times())).reduceByKey((x,y)->x+y)
                                         .mapPartitions(data->{

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 TagFilter {
+public class TagCouponFilter implements BaseTagFilter {
 
     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 TagFilter {
+public class TagCurrentCouponNumFilter implements BaseTagFilter {
 
     private DataSourceSharding dataSourceSharding=new DataSourceSharding(AppEnvUtil.MARKETING_SHARDING, ConstantUtil.TAB_COUPON_LOG);
     private static TagCurrentCouponNumFilter instance;

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

@@ -46,7 +46,7 @@ public class TagFirstConsumeCommodityFilter extends AbstractTagConsumRecordFilte
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeCommodityRequest commodityRequest=(TagConsumeCommodityRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,commodityRequest);
         JavaPairRDD<Long,Long>orderRdd= consumeRecordRDD.filter(data-> StringUtils.isNotEmpty(data.getReceipts_time()))
@@ -60,7 +60,7 @@ public class TagFirstConsumeCommodityFilter extends AbstractTagConsumRecordFilte
                                                                 }
                                                             })
                                                             .mapToPair(data->Tuple2.apply(data._2().getVirtual_id(),data._1()));
-        JavaPairRDD<Long,Long> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD()
+        JavaPairRDD<Long,Long> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEntId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD()
                                                                 .filter(data->{
                                                                     if(StringUtils.isNotEmpty(data.getSku_code())
                                                                         &&commodityRequest.getSkuCodeList().contains(data.getSku_code())){

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

@@ -41,7 +41,7 @@ public class TagFirstConsumeTimeFilter extends AbstractTagConsumRecordFilter {
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeTimeRequest consumeTimeRequest=(TagConsumeTimeRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId)
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId)
                                                                         .filter(new Column("is_eff_order").equalTo(1)), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeTimeRequest);
 

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

@@ -37,7 +37,7 @@ public class TagFirstConsumptionChannelFilter extends AbstractTagConsumRecordFil
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeChannelRequest channelRequest=(TagConsumeChannelRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,channelRequest);
 

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

@@ -39,7 +39,7 @@ public class TagFirstConsumptionMoneyFilter extends AbstractTagConsumRecordFilte
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);
 

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

@@ -42,7 +42,7 @@ public class TagFirstOfflineConsumptionStoreFilter extends AbstractTagConsumReco
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()==1

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

@@ -41,7 +41,7 @@ public class TagFirstOnlineConsumptionStoreFilter extends AbstractTagConsumRecor
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()!=1

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

@@ -46,10 +46,10 @@ public class TagHistoryConsumeCommodityFilter extends AbstractTagConsumRecordFil
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeCommodityRequest commodityRequest=(TagConsumeCommodityRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,commodityRequest);
-        JavaRDD<TrdVirtualOrderItemBean> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderItemBean> orderItemRDD=MysqlRddManager.getPojoFromDataset(dataSourceHiveOrderItem.getDatasetByEntId(enterpriseId),TrdVirtualOrderItemBean.class).javaRDD();
         JavaRDD<Long>ecuRdd=consumeRecordRDD.mapToPair(data->Tuple2.apply(data.getVirtual_id(),data.getEcu_id()))
                                                 .leftOuterJoin(orderItemRDD.mapToPair(data->Tuple2.apply(data.getVirtual_order_id(),data.getSku_code()))
                                                                             .filter(data->StringUtils.isNotEmpty(data._2())

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

@@ -40,7 +40,7 @@ public class TagHistoryConsumeTotalFilter extends AbstractTagConsumRecordFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);
 

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

@@ -40,7 +40,7 @@ public class TagHistoryOfflineConsumptionStoreFilter extends AbstractTagConsumRe
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()==1

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

@@ -40,7 +40,7 @@ public class TagHistoryOnlineConsumptionStoreFilter extends AbstractTagConsumRec
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()!=1

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

@@ -46,7 +46,7 @@ public class TagLatelyConsumeCommodityFilter extends AbstractTagConsumRecordFilt
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeCommodityRequest commodityRequest=(TagConsumeCommodityRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,commodityRequest);
         JavaPairRDD<Long,Long>orderRdd= consumeRecordRDD.filter(data-> StringUtils.isNotEmpty(data.getReceipts_time()))

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

@@ -40,7 +40,7 @@ public class TagLatelyConsumeTimeFilter extends AbstractTagConsumRecordFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeTimeRequest consumeTimeRequest=(TagConsumeTimeRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId)
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId)
                                                         .filter(new Column("is_eff_order").equalTo(1)), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeTimeRequest);
 

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

@@ -40,7 +40,7 @@ public class TagLatelyConsumptionChannelFilter extends AbstractTagConsumRecordFi
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeChannelRequest channelRequest=(TagConsumeChannelRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,channelRequest);
 

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

@@ -42,7 +42,7 @@ public class TagLatelyConsumptionMoneyFilter  extends AbstractTagConsumRecordFil
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);
 

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

@@ -41,7 +41,7 @@ public class TagLatelyOnlineConsumptionStoreFilter extends AbstractTagConsumReco
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()!=1

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

@@ -40,7 +40,7 @@ public class TagLowestSingleConsumptionMoneyFilter extends AbstractTagConsumReco
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);
 

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

@@ -43,7 +43,7 @@ public class TagOfflineConsumptionStoreFilter extends AbstractTagConsumRecordFil
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
         storeRequest.setYearMonthDayType(YearMonthDayType.DAY);
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()==1

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

@@ -38,7 +38,7 @@ public class TagOnlineConsumptionStoreFilter extends AbstractTagConsumRecordFilt
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
         storeRequest.setYearMonthDayType(YearMonthDayType.DAY);
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()!=1

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

@@ -37,7 +37,7 @@ public class TagPerCustomerTransactionFilter extends AbstractTagConsumFilter{
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeRequest consumeRequest=(TagConsumeRequest)request;
-        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
+        JavaRDD<TrdEcuSalesLabelBean> consumeRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdEcuSalesLabelBean.class).javaRDD();
         consumeRDD=statisticsTypeHandle(consumeRDD,consumeRequest);
         JavaRDD<Long>ecuRdd=consumeRDD.mapToPair(data-> Tuple2.apply(data.getEcu_id(),data)).groupByKey()
                                           .mapPartitions(data->{

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

@@ -40,7 +40,7 @@ public class TagRecentlyOfflineConsumptionStoreFilter extends AbstractTagConsumR
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
 
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,storeRequest);
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data-> data.getOrder_channel_code()==1

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

@@ -39,7 +39,7 @@ public class TagTopSingleConsumptionMoneyFilter  extends AbstractTagConsumRecord
     @Override
     public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
         TagConsumeAmountRequest consumeAmountRequest=(TagConsumeAmountRequest)request;
-        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEnterpriseId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
+        JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHive.getDatasetByEntId(enterpriseId), TrdVirtualOrderBean.class).javaRDD();
         consumeRecordRDD=statisticsTypeHandle(consumeRecordRDD,consumeAmountRequest);
         int configStatus= CommonUtil.getConfigStatus(enterpriseId);
 

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

@@ -23,8 +23,8 @@ public class TagFilterFactory {
     }
     private TagFilterFactory(){}
 
-    public TagFilter getTagFilter(TagConditionDTO conditionDTO){
-        TagFilter tagFilter=null;
+    public BaseTagFilter getTagFilter(TagConditionDTO conditionDTO){
+        BaseTagFilter tagFilter=null;
         if(StringUtils.isNotEmpty(conditionDTO.getTagEsFieldName())){
             switch (conditionDTO.getTagEsFieldName()) {
                 //积分信息

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

@@ -129,7 +129,7 @@ public class TagProcessManager {
 //        }
 
         Map<Integer, List<TagProcessEntity>> tagGroupByEnterpriseMap = new HashMap<>();
-        Map<Long, TagFilter> tagIdToFilterMap = new HashMap();
+        Map<Long, BaseTagFilter> tagIdToFilterMap = new HashMap();
 
         for(SceneCrowdDTO sceneCrowdDTO:sceneCrowdDTOList){
             if(sceneCrowdDTO.getReal_Time()==2){
@@ -143,7 +143,7 @@ public class TagProcessManager {
 
                     entity.tagList=conditionGroupDTOS.get(i).getConditionInfos();
                     for(TagConditionDTO conditionDTO:entity.tagList){//将tag同filter进行映射
-                        TagFilter tagFilter=TagFilterFactory.getInstance().getTagFilter(conditionDTO);
+                        BaseTagFilter tagFilter=TagFilterFactory.getInstance().getTagFilter(conditionDTO);
 
                         if(null!=tagFilter){
                             tagIdToFilterMap.put(conditionDTO.getTagId(),tagFilter);
@@ -198,7 +198,7 @@ public class TagProcessManager {
             for(TagProcessEntity entity:enterpriseTagEntry.getValue()){
                 for(TagConditionDTO conditionDTO:entity.tagList){
                     if(tagIdToFilterMap.containsKey(conditionDTO.getTagId())){
-                        TagFilter tagFilter=tagIdToFilterMap.get(conditionDTO.getTagId());
+                        BaseTagFilter tagFilter=tagIdToFilterMap.get(conditionDTO.getTagId());
                         AbstractFilterRequest filterRequest=TagValueParser.parseFilterValue(conditionDTO);
 
                         final String groupId = entity.tagGroupId + "_" + conditionDTO.getTagId() + "_" + entity.level;