twitter/scalding — pull request #761 by galarragas (2014-01-28 13:46:49+00:00)
Title: BDD DSL to test pipe transformations 

This commit allows to write unit tests in BDD style for functions working on one or more input pipes and generating a resulting pipe. It allows to tests parts of a job in isolation without using checkpoints. It is the porting of the content of my project https://github.com/galarragas/ScaldingUnit. After having used it for a while I think is stable enough to be merged into Scalding.

An example of code to test the addUserInfo function:

```scala
class SampleJobPipeTransformationsSpec extends FlatSpec with ShouldMatchers with TupleConversions with ScalaTestScaldingSupport {
"A sample job pipe transformation" should "add user info" in {
Given {
List(("2013/02/11", 1000002l, 1l)) withSchema EVENT_COUNT_SCHEMA
} And {
List( (1000002l, "stefano@email.com", "10 Downing St. London") ) withSchema USER_DATA_SCHEMA
} When {
(eventCount: RichPipe, userData: RichPipe) => eventCount.addUserInfo(userData)
} Then {
buffer: mutable.Buffer[(String, Long, String, String, Long)] =>
buffer.toList shouldEqual List( ("2013/02/11", 1000002l, "stefano@email.com", "10 Downing St. London", 1l) )
}
}
}
```

Where the addUserInfo function is defined in this way:

```scala
package object SampleJobPipeTransformations {

val INPUT_SCHEMA = List('date, 'userid, 'url)
val WITH_DAY_SCHEMA = List('date, 'userid, 'url, 'day)
val EVENT_COUNT_SCHEMA = List('day, 'userid, 'event_count)
val OUTPUT_SCHEMA = List('day, 'userid, 'email, 'address, 'event_count)

val USER_DATA_SCHEMA = List('userid, 'email, 'address)

val INPUT_DATE_PATTERN: String = "dd/MM/yyyy HH:mm:ss"

implicit def wrapPipe(pipe: Pipe): SampleJobPipeTransformationsWrapper = new SampleJobPipeTransformationsWrapper(new RichPipe(pipe))
implicit class SampleJobPipeTransformationsWrapper(val self: RichPipe) extends PipeOperations {

/**
* Input schema: INPUT_SCHEMA
* Output schema: WITH_DAY_SCHEMA
* @return
*/
def addDayColumn = self.map('date -> 'day) {
date: String => DateTimeFormat.forPattern(INPUT_DATE_PATTERN).parseDateTime(date).toString("yyyy/MM/dd");
}

/**
* Input schema: WITH_DAY_SCHEMA
* Output schema: EVENT_COUNT_SCHEMA
* @return
*/
def countUserEventsPerDay = self.groupBy(('day, 'userid)) { _.size('event_count) }

/**
* Joins with userData to add email and address
*
* Input schema: WITH_DAY_SCHEMA
* User data schema: USER_DATA_SCHEMA
* Output schema: OUTPUT_SCHEMA
*/
def addUserInfo(userData: Pipe) = self.joinWithLarger('userid -> 'userid, userData).project(OUTPUT_SCHEMA)
}
}
```

This will allow to write Jobs as aggregation of independently tested functions as this:

```scala
import SampleJobPipeTransformations._

class SampleJob(args: Args) extends Job(args) {

Osv(args("eventsPath")).read
.addDayColumn
.countUserEventsPerDay
.addUserInfo(Osv(args("userInfoPath")).read)
.write(Tsv(args("outputPath")))
}
```