会员标签4.0

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

@@ -65,7 +65,7 @@ public class TagFirstOfflineConsumptionStoreFilter extends AbstractTagConsumReco
                                                     }else{
                                                         return y;
                                                     }
-                                                }).filter(data->storeRequest.getStoreList().contains(data._2()._2()))
+                                                }).filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2()._2())))
                                                 .map(data->data._1());
 
         return ecuRdd;

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

@@ -65,7 +65,7 @@ public class TagFirstOnlineConsumptionStoreFilter extends AbstractTagConsumRecor
                                                     }else{
                                                         return y;
                                                     }
-                                                }).filter(data->storeRequest.getStoreList().contains(data._2()._2()))
+                                                }).filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2()._2())))
                                                 .map(data->data._1());
         return ecuRdd;
     }

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

@@ -45,7 +45,7 @@ public class TagHistoryOfflineConsumptionStoreFilter extends AbstractTagConsumRe
         TagConsumeStoreRequest storeRequest=(TagConsumeStoreRequest)request;
         JavaRDD<TrdVirtualOrderBean>consumeRecordRDD= MysqlRddManager.getPojoFromDataset(dataSourceHiveOrder.getDatasetByEntId(enterpriseId),TrdVirtualOrderBean.class).javaRDD();
 
-        JavaRDD<Row>virtualOrderItemRdd=dataSourceHiveOrderItem.getDatasetByEntId(enterpriseId).select("virtual_order_id","ent_brand_id").javaRDD();
+        JavaRDD<Row>virtualOrderItemRdd=dataSourceHiveOrderItem.getDatasetByEntId(enterpriseId).select("virtual_order_id","ent_brand_id").toJavaRDD();
 
         JavaRDD<Tuple2<TrdVirtualOrderBean,Optional<Iterable<String>>>>orderRdd=consumeRecordRDD.mapToPair(data->Tuple2.apply(data.getVirtual_id(),data))
                                                                                 .leftOuterJoin(virtualOrderItemRdd.mapToPair(row->Tuple2.apply(row.getLong(0),row.getString(1))).groupByKey())
@@ -55,7 +55,7 @@ public class TagHistoryOfflineConsumptionStoreFilter extends AbstractTagConsumRe
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()==1
                                                               && null!=data.getStore_info_id())
                                                 .mapToPair(data-> Tuple2.apply(data.getEcu_id(),data.getStore_info_id()))
-                                                .filter(data->storeRequest.getStoreList().contains(data._2()))
+                                                .filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2())))
                                                 .reduceByKey((x,y)->x)
                                                 .map(data->data._1());
 

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

@@ -55,7 +55,7 @@ public class TagHistoryOnlineConsumptionStoreFilter extends AbstractTagConsumRec
         JavaRDD<Long>ecuRdd=consumeRecordRDD.filter(data->data.getOrder_channel_code()!=1
                                                 && null!=data.getShop_id())
                                                 .mapToPair(data-> Tuple2.apply(data.getEcu_id(),data.getShop_id()))
-                                                .filter(data->storeRequest.getStoreList().contains(data._2()))
+                                                .filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2())))
                                                 .reduceByKey((x,y)->x)
                                                 .map(data->data._1());
         return ecuRdd;

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

@@ -65,7 +65,7 @@ public class TagLatelyOnlineConsumptionStoreFilter extends AbstractTagConsumReco
                                                     }else{
                                                         return y;
                                                     }
-                                                }).filter(data->storeRequest.getStoreList().contains(data._2()._2()))
+                                                }).filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2()._2())))
                                                 .map(data->data._1());
         return ecuRdd;
     }

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

@@ -60,7 +60,7 @@ public class TagOfflineConsumptionStoreFilter extends AbstractTagConsumRecordFil
                                                 && StringUtils.isNotEmpty(data.getReceipts_time()))
                                                 .filter(data->checkTime(storeRequest, DateUtil.strToDate(data.getReceipts_time(),DateUtil.FORMAT_DATETIME_19).getTime()))
                                                 .mapToPair(data-> Tuple2.apply(data.getEcu_id(),data.getStore_info_id()))
-                                                .filter(data->storeRequest.getStoreList().contains(data._2()))
+                                                .filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2())))
                                                 .reduceByKey((x,y)->x)
                                                 .map(data->data._1());
         return ecuRdd;

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

@@ -55,7 +55,7 @@ public class TagOnlineConsumptionStoreFilter extends AbstractTagConsumRecordFilt
                                                 && null!=data.getShop_id())
                                                 .filter(data->checkTime(storeRequest, DateUtil.strToDate(data.getReceipts_time(),DateUtil.FORMAT_DATETIME_19).getTime()))
                                                 .mapToPair(data-> Tuple2.apply(data.getEcu_id(),Tuple2.apply(data.getReceipts_time(),data.getShop_id())))
-                                                .filter(data->storeRequest.getStoreList().contains(data._2()._2()))
+                                                .filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2()._2())))
                                                 .reduceByKey((x,y)->x)
                                                 .map(data->data._1());
         return ecuRdd;

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

@@ -63,7 +63,7 @@ public class TagRecentlyOfflineConsumptionStoreFilter extends AbstractTagConsumR
                                                                 }else{
                                                                     return y;
                                                                 }
-                                                            }).filter(data->storeRequest.getStoreList().contains(data._2()._2()))
+                                                            }).filter(data->storeRequest.getStoreList().contains(String.valueOf(data._2()._2())))
                                                             .map(data->data._1());
 
         return ecuRdd;