twitter/scalding — pull request #139 by xianxu (2012-07-25 03:03:30+00:00)
Title: Side effect in RichPipe

This adds a way to access expensive nonserializable object in a map/flatMap function. It's needed for scalding to use certain library that requires expensive set up. This usage is captured with construct: using (state) map { ... } and using (state) flatMap {...}. The result of such construct is a pipe and can be further chained.

It can also be used to introduce side effect, such as writing data directly into Memcache. In this case, the pattern is: using (state) foreach { ... }. This will end the pipe and no additional stage should be appended.

In both cases, much like existing map/flatMap, the state supplied in using clause is passed to map/flatMap/foreach as first argument. The state's release() function's automatically called when work's done.

Some example code:

def createState = new Stateful {
val memcacheClient = ClientBuilder()...build()
override def release() {
memcacheClient.release()
}
}

pipe
.using (createClient)
.foreach { case (state, (key, value)) =>
state.memcacheClient.add(key, value)
}