Kinesis record reading: thawing non-nippy data

[paxan]

I have Kinesis records that contain Snappy-encoded blocks such that when a process that decodes each block yields a sequence of strings like this:

"{:foo 42} "
"{:bar 3.14} "
...
"{:qaz [1 2 3 4]} "

So each string is an EDN value (a basic Clojure literal).

Amazonica seems to assume that data must be nippy-serialized. So when I try:

(get-records :shard-iterator (get-shard-iterator "my-stream"
                                                 "shardId-000000000003"
                                                 "TRIM_HORIZON"))

I get:

CompilerException java.lang.Exception: Thaw failed: Uncompressed data?, compiling:(form-init1883914083661306631.clj:1:9)

Perhaps I am missing it, but I don't see a simple way to tell Amazonica not to do that and instead give me raw bytes, which I could decompress and decode however I like.
So, is there a way to do that?

So far I see that you have unwrap function that is used directly and indirectly by get-records and by processor-factory, respectively. Would be wonderful if I could supply my own version of unwrap, instead of nippy-thawing version of unwrap.

[mcohen01]

Yeah, I guess I assumed if you were using clojure to read from kinesis that you were putting the records into kinesis from clojure, in which case they'd be nippy frozen. Bad assumption. I'll parameterize the unwrap function.

[mcohen01]

The fix version is on Clojars. If you want control over serialization and deserialization, you can pass a nio.ByteBuffer to calls to (put-record), in which case it won't be nippy/frozen, and supply a :deserializer function to (get-records) or (worker!) calls, which will the nio.ByteBuffer as their only arg and can return anything, which will be available under the :data key, same as before. I think that should work for your case, no?

[paxan]

Michael, thank you. This sounds great! It is not likely I will have a chance to test it until Feb 3, but it does seem exactly what we needed in our scenario.

On Sat, Jan 25, 2014 at 1:36 PM, Michael Cohen notifications@github.com
wrote:

The fix version is on Clojars. If you want control over serialization and deserialization, you can pass a nio.ByteBuffer to calls to (put-record), in which case it won't be nippy/frozen, and supply a :deserializer function to (get-records) or (worker!) calls, which will the nio.ByteBuffer as their only arg and can return anything, which will be available under the :data key, same as before. I think that should work for your case, no?

Reply to this email directly or view it on GitHub:
#36 (comment)

[paxan]

Thank you, @mcohen01

Fixed by c50213f.