ad feature to redis

src/main/java/examples/dataloader/AdRedisFeatureConstructor.java

@@ -51,7 +51,7 @@ public class AdRedisFeatureConstructor {
 
     public static UserAdFeature constructUserFeature(Record record) {
         UserAdFeature userFeature = new UserAdFeature();
-        userFeature.setMid(record.get("mids").toString());
+        userFeature.setMid(record.getString("mids"));
 
         // 1day features
         AdActionFeature userAd1dayActionFeature = new AdActionFeature();
@@ -63,7 +63,6 @@ public class AdRedisFeatureConstructor {
         userFeature.setDay1_cnt_features(userAd1dayActionFeature);
 
 
-
         // 3day features
         AdActionFeature userAd3dayActionFeature = new AdActionFeature();
         userAd1dayActionFeature.setAdView(record.getString("ad_view_3day"));
@@ -84,7 +83,6 @@ public class AdRedisFeatureConstructor {
         userFeature.setDay7_cnt_features(userAd7dayActionFeature);
 
 
-
         // 3month features
         AdActionFeature userAd3MonthActionFeature = new AdActionFeature();
         userAd3MonthActionFeature.setAdView(record.getString("ad_view_3month"));
@@ -102,6 +100,7 @@ public class AdRedisFeatureConstructor {
     public static AdItemFeature constructItemFeature(Record record) {
         AdItemFeature itemFeature = new AdItemFeature();
         itemFeature.setAdId(record.getString("adid"));
+        // itemFeature.setAdCode(record.getString("adcode"));
         itemFeature.setCampaignId(record.getString("campaignid"));
         itemFeature.setAdvertiserId(record.getString("advertiserid"));
         itemFeature.setCreativeId(record.getString("creativeid"));

src/main/java/examples/dataloader/AdSampleConstructor.java

@@ -93,6 +93,7 @@ public class AdSampleConstructor {
 
 
         itemFeature.setAdId(record.getString("adid"));
+        // itemFeature.setAdCode(record.getString("adcode"));
         itemFeature.setAdvertiserId(record.getString("advertiserid"));
         itemFeature.setCampaignId(record.getString("campaignid"));
         itemFeature.setCreativeId(record.getString("creativeid"));

src/main/java/examples/sparksql/SparkAdFeaToRedisLoader.java

@@ -16,7 +16,9 @@ import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 
@@ -42,34 +44,40 @@ public class SparkAdFeaToRedisLoader {
     }
 
 
-    public static void loadFeatureToRedis(RedisTemplate<String, String> redisTemplate, String line) {
+    public static void loadFeatureToRedis(RedisTemplate<String, String> redisTemplate, List<String> line) {
         Map<String, String> redisFormat = new HashMap<String, String>();
-        String key = line.split("\t")[0];
-        String value = line.split("\t")[1];
+        String key = line.get(0);
+        String value = line.get(1);
         redisFormat.put(key, value);
         redisTemplate.opsForValue().multiSet(redisFormat);
-
     }
 
 
-    static class RecordsToAdRedisKV implements Function2<Record, TableSchema, String> {
+
+    static class RecordsToAdRedisKV implements Function2<Record, TableSchema, List<String>> {
         @Override
-        public String call(Record record, TableSchema schema) throws Exception {
+        public List<String> call(Record record, TableSchema schema) throws Exception {
             AdItemFeature adItemFeature = AdRedisFeatureConstructor.constructItemFeature(record);
             String key = String.format(adKeyFormat, adItemFeature.getKey());
             String value = adItemFeature.getValue();
-            return key + "\t" + value;
+            List<String> kv = new ArrayList<String>();
+            kv.add(key);
+            kv.add(value);
+            return kv;
         }
     }
 
 
-    static class RecordsToUserRedisKV implements Function2<Record, TableSchema, String> {
+    static class RecordsToUserRedisKV implements Function2<Record, TableSchema, List<String>> {
         @Override
-        public String call(Record record, TableSchema schema) throws Exception {
+        public List<String> call(Record record, TableSchema schema) throws Exception {
             UserAdFeature userFeature = AdRedisFeatureConstructor.constructUserFeature(record);
             String key = String.format(userKeyFormat, userFeature.getKey());
             String value = userFeature.getValue();
-            return key + "\t" + value;
+            List<String> kv = new ArrayList<String>();
+            kv.add(key);
+            kv.add(value);
+            return kv;
         }
     }
 
@@ -93,7 +101,7 @@ public class SparkAdFeaToRedisLoader {
 
 
         // load Ad features
-        JavaRDD<String> readAdData = odpsOps.readTableWithJava(project, tableAdInfo, partition, new RecordsToAdRedisKV(), Integer.valueOf(10));
+        JavaRDD<List<String>> readAdData = odpsOps.readTableWithJava(project, tableAdInfo, partition, new RecordsToAdRedisKV(), Integer.valueOf(10));
         readAdData.foreachPartition(
                 rowIterator -> {
                     RedisTemplate<String, String> redisTemplate = buildRedisTemplate();
@@ -103,8 +111,8 @@ public class SparkAdFeaToRedisLoader {
 
 
         // load user features
-        JavaRDD<String> readUserData = odpsOps.readTableWithJava(project, tableUserInfo, partition, new RecordsToUserRedisKV(), Integer.valueOf(30));
-        readUserData.foreachPartition(
+        JavaRDD<List<String>> readUserData = odpsOps.readTableWithJava(project, tableUserInfo, partition, new RecordsToUserRedisKV(), Integer.valueOf(50));
+        readUserData.repartition(50).foreachPartition(
                 rowIterator -> {
                     RedisTemplate<String, String> redisTemplate = buildRedisTemplate();
                     rowIterator.forEachRemaining(line -> loadFeatureToRedis(redisTemplate, line));