Add new zset mode

README.md

@@ -19,12 +19,14 @@ Use:
 
 ## Bulkloading strategy
 
-RedisStorer runs in four modes: kv, set, hash and list (specified as the first argument to RedisStorer). If no mode is specified, kv is the default.
+RedisStorer runs in four modes: kv, set, zset, hash and list (specified as the first argument to RedisStorer). If no mode is specified, kv is the default.
 
 In kv mode, it takes the first field of the stored tuple as the key, and the second field as the value, and issues [SET key value](http://redis.io/commands/set). Any further fields are ignored.
 
 In set mode, it takes the first field of the stored tuple as the key, and issues [SADD key value](http://redis.io/commands/sadd) once for each subsequent field value in the tuple.
 
+In set mode, it takes the first field of the stored tuple as the key, and issues [ZADD key value](http://redis.io/commands/zadd) once for each subsequent two field values in the tuple. The first field value will be interpreted as a Double typed score, the second will be interpreted as the value itself.
+
 In hash mode, it takes the first field of the stored tuple as the key, and issues [HSET key fieldname value](http://redis.io/commands/hset) once for each subsequent field value, using the same key for each, and taking the fieldname from the tuple's schema fieldnames. This means that it will fail unless the stored tuple has a schema with named fields.
 
-In list mode, it takes the first field of the stored tuple as the key, and issues [LPUSH key value](http://redis.io/commands/lpush) once for each subsequent field value in the tuple.
+In list mode, it takes the first field of the stored tuple as the key, and issues [LPUSH key value](http://redis.io/commands/lpush) once for each subsequent field value in the tuple.

src/com/hackdiary/pig/RedisStorer.java

@@ -75,6 +75,37 @@ else if(_mode.equals("set")) {
         }
         p.execute();
       }
+      else if(_mode.equals("zset")) {
+        int idx = 0;
+        Pipeline p = _jedis.pipelined();
+        for(Object o : values) {
+          if(idx != 0 && o != null) {
+            switch (DataType.findType(o)) {
+              case DataType.TUPLE:
+              case DataType.BAG:
+                {
+                  String[] a = new String[2];
+                  int index = 0;
+                  for (Object o2 : (Iterable)o) {
+                    a[index++] = o2.toString();
+                    if (index == 2) {
+                      p.zadd(key, Double.parseDouble(a[1]), a[0]);
+                      index = 0;
+                    }
+                  }
+                }
+                break;
+              default:
+                /* If we are given any other type, what does that even mean with a zset?
+                 * Arbitrarily decide to set to 0.0 score. Need input from maintainers. */
+                p.zadd(key, 0.0, o.toString());
+                break;
+            }
+          }
+          idx++;
+        }
+        p.execute();
+      }
       else if(_mode.equals("hash")) {
         UDFContext context  = UDFContext.getUDFContext();
         Properties property = context.getUDFProperties(ResourceSchema.class);