Riak-Java cannot deserialize domain objects from MapReduce query in Scala?

Question

Performing a MapReduce query on a simple bucket. For some reason, I am getting an exception from Jackson:

Caused by: org.codehaus.jackson.map.JsonMappingException: Can not deserialize instance of com.threetierlogic.AccountService.models.User out of START_ARRAY token
 at [Source: java.io.StringReader@39494ff0; line: 1, column: 2]
        at org.codehaus.jackson.map.JsonMappingException.from(JsonMappingException.java:163)
        at org.codehaus.jackson.map.deser.StdDeserializationContext.mappingException(StdDeserializationContext.java:219)
        at org.codehaus.jackson.map.deser.StdDeserializationContext.mappingException(StdDeserializationContext.java:212)
        ... 179 more

Here's the MapReduce Query I am performing:

DB.client.mapReduce(bucketName)
    .addReducePhase(NamedErlangFunction.REDUCE_IDENTITY)
    .execute().getResult(classOf[User])

Now if I call the JSON using getResultRaw() it returns a string of JSON that contains no values, only keys:

[["accounts-user","8f0bb6e41592690d701225e263807a5e"],["accounts-user","2687cf9444013c45ba2637e9f6d3d3ad"],["accounts-user","3507e2e1f3d2818fdd276214d594c8e"],["accounts-user","fd186b0293ab7eb737f8b66e353fe4a6"],["accounts-user","bf6ce6bca0f642abfe74f2e2281e676c"],["accounts-user","b58d356551a8df6d3bbaf65e577f4b12"],["accounts-user","8126d599d259fd43f701c90787096049"],["accounts-user","33b9ae3befb23b7030b609158bb762d"],["accounts-user","770a897d5ce8c8e118ae121fc01f4c80"],["accounts-user","edae605390c35256b5df055f5574734d"],["accounts-user","ef19ad34a2be4ab8de111d1590a8768b"],["accounts-user","89a9f29ac937595038d37169f9ba7c8"],["accounts-user","85be26f43f7bb74eefa7683dcc74c555"]]

So what am I overlooking here? Do I need to invoke some sort of domain converter or is the problem in the MapReduce query itself?

Edit

If it helps, here's how I store IRiakObjects in Riak:

def store(o: User) = bucket.store(o).withConverter(new UserConverter(bucketName)).execute()

Answer 1

First problem as noted in the comments is that REDUCE_IDENTITY returns a list of bucket/key pairs which isn't what you want. Second problem is ... you can't go straight to your POJO from mapreduce if you're storing something other than JSON. Third (sorta) problem is ... Riak mapreduce is really not made for binary values.

The following uses the KryoPersonConverter and Person class from the example in the cookbook that demonstrates how to use Kryo in a custom converter.

I'm going to break to code sample up and inline my comments:

public class App 
{
    public static void main( String[] args ) throws RiakException
    {
        List<Person> personList = new ArrayList<Person>();

        Person p = new Person("Bob","111 Elm Street","555-1212");
        personList.add(p);
        p = new Person("Jenny","122 Spruce Lane","867-5309");
        personList.add(p);
        p = new Person("Steve","333 Oak place","555-1111");
        personList.add(p);


        IRiakClient client = RiakFactory.pbcClient();
        Bucket b = client.fetchBucket("people").execute();
        KryoPersonConverter converter = new KryoPersonConverter("people");

        for (Person p2 : personList)
        {
            b.store(p2).withConverter(converter).execute();
        }

        p = new Person();
        p.setName("Jenny");
        p = b.fetch(p).withConverter(converter).execute();
        assert(p.getPhone().equals("867-5309")); // I got your number

Everything up to now? A-OK! We've stored a POJO in Riak after having used Kryo to serialize it, and retrieved it.

        MapReduceResult result = client.mapReduce("people")
                                       .addMapPhase(NamedErlangFunction.MAP_OBJECT_VALUE)
                                       .execute();

        System.out.println(result.getResultRaw());

And here we see the problem, as the output of that println() is:

["\u0001\u0001\u000e111 Elm Street\u0001\u0003Bob\u0001\b555-1212","\u0001\u0001\r333 Oak place\u0001\u0005Steve\u0001\b555-1111","\u0001\u0001\u000f122 Spruce Lane\u0001\u0005Jenny\u0001\b867-5309"]

Unfortunately, mapreduce in Riak is really meant to be used with JSON data (or just plain strings) when talking about stored data. We have a JSON array containing JSON strings of the bytes we stored.

To work with that, you'd have to get the Strings as a Collection, then convert the bytes using Kryo.

        Collection<String> objects = result.getResult(String.class);


        Kryo kryo = new Kryo();
        kryo.register(Person.class);
        ObjectBuffer buffer = new ObjectBuffer(kryo);

        for (String s : objects)
        {
            Person p3 = buffer.readObject(s.getBytes(), Person.class);
            System.out.println(p3.getName() + " " + p3.getPhone());
        }

        client.shutdown();     
    }
}

And you'd get the output:

Bob 555-1212
Steve 555-1111
Jenny 867-5309