会员标签4.0

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

@@ -2,10 +2,16 @@ package com.gic.spark.filter;
 
 import com.gic.spark.datasource.entity.DataSourceEntity;
 import com.gic.spark.datasource.entity.DataSourceSharding;
+import com.gic.spark.datasource.mysql.MysqlRddManager;
+import com.gic.spark.entity.bean.TagIntegralBean;
 import com.gic.spark.entity.request.AbstractFilterRequest;
+import com.gic.spark.entity.request.TagIntegralRequest;
 import com.gic.spark.util.AppEnvUtil;
 import com.gic.spark.util.ConstantUtil;
 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;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -36,7 +42,11 @@ public class TagAboutExpireIntegralFilter implements BaseTagFilter {
     }
 
     @Override
-    public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest filterOption) {
+    public JavaRDD<Long> filterValidMember(Integer enterpriseId, AbstractFilterRequest request) {
+        TagIntegralRequest integralRequest=(TagIntegralRequest)request;
+        Dataset<Row> dataset=dataSourceSharding.getDatasetByEnterpriseId(enterpriseId)
+                                                .filter(new Column("create_Time").isNotNull());
+        JavaRDD<TagIntegralBean>integralRDD= MysqlRddManager.getPojoFromDataset(dataset,TagIntegralBean.class).javaRDD();
 
 
         return null;