A micro-test framework for scalding pipes to make sure you don't get burnt

The thermometer library has a few goals:

• Be explicit in expected outcomes, whilst not being concerned with irrelevant details.
• Provide exceptional feedback in the face of failure:
  • Good error messages.
  • Clear mapping into actual files on disk where appropriate.
• To allow testing of end-to-end pipelines which is impossible with JobTest.
• To just work (no race-conditions, data to clean-up etc...), and work fast.

Thermometer tests can be declared in two ways, as facts or as traditional specs2 checks. The facts api should be preferred, it generally provides better contextual error messages and composition for lower effort.

Scaladoc

See https://commbank.github.io/thermometer/index.html.

Import everything:

import au.com.cba.omnia.thermometer.core._, Thermometer._
import au.com.cba.omnia.thermometer.fact.PathFactoids._
import com.twitter.scalding._

Then create a spec that extends ThermometerSpec. This sets up appropriate scalding, cascading and hadoop related things as well as ensuring that specs2 is run in a way that won't break hadoop.

Facts can be asserted on cascading Pipe objects or scalding TypedPipe objects.

To verify some pipeline, you add a withFacts call. For example:

  def pipeline =
    ThermometerTestSource(List("hello", "world"))
      .map(c => (c, "really" + c + "!"))
      .write(TypedPsv[(String, String)]("output"))
      .withFacts(
        "cars" </> "_ERROR"      ==> missing
      , "cars" </> "_SUCCESS"    ==> exists
      , "cars" </> "part-00000"  ==> (exists, count(data.size))
      )

Breaking this down, withFacts takes a sequence of Facts, these can be construted in a number of ways, the most supported form are PathFacts, which are built using the ==> operation added to hdfs Paths and Strings. The right hand side of ==> specifies a sequences of facts that should hold true given the specified path.

Thermometer expectations allow you to fall back to specs2, this may be because of missing functionality from the facts api, or for optimisation of special cases.

To verify some pipeline, you add a withExpectations call. For example:

  def pipeline =
    ThermometerTestSource(List("hello", "world"))
      .map(c => (c, "really" + c + "!"))
      .write(TypedPsv[(String, String)]("output"))
      .withExpectations(context => {
         context.exists("output" </> "_SUCCESS") must beTrue
         context.lines("output" </> "part-*").toSet must_== Set(
           "hello" -> "really hello!",
           "world" -> "really world!"
         )
      })

Breaking this down, withExpectations takes a function Context => Unit. Context is a primitive (unsafe) API over hdfs operations that will allow you to make assertions. The Context handles unexpected failures by failing the test with a nice error message, but there is no way to do manual error handling at this point.

A ThermometerSource is a thin wrapper around an in-memory scalding source that is specialized so that it can be immediately treated as a TypedPipe without corner cases (and better inference).

Usage:

  def pipeline =
    ThermometerSource(List("hello", "world"))            // : TypedPipe[String]
      .map(c => (c, "really" + c + "!"))
      .write(TypedPsv[(String, String)]("output"))

The hackery that thermometer uses to handle the mutable, global, implicit state that scalding uses (yes shake your head now). Needs to be reset for each run. To do this use an isolate {} block inside the property.

For example:

  def pipeline = prop((data: List[String]) => isolate {
    ThermometerSource(data)
      .map(c => (c, "really " + c + "!"))
      .write(TypedPsv[(String, String)]("output"))
      .withFacts(
      , "output" </> "_SUCCESS"    ==> exists
      )
  })

It is often useful to use one spec as the input to another (for example, you want to write then read).

To do this use withDependency.

For example if you were testing TypedPsv/TypedCsv something like this would work:

  def write =
    ThermometerSource(List("hello", "world"))
      .write(TypedPsv[String]("output.psv"))
      .withFacts(
        "output.psv" </> "_SUCCESS"   ==> exists
      )

  def read = withDependency(write) {
    TypedPsv[String]("output.psv")
      .write(TypedCsv("output.csv"))
      .withFacts(
        "customers.csv" </> "_SUCCESS"   ==> exists
      )
  }

Mix in the HiveSupport trait in import au.com.cba.omnia.thermometer.hive.HiveSupport to add support for hive and, in particular, set up a separate warehouse directory and metadata database per test and provide the right HiveConf.

• Built-in support for more facts
  • Support for streaming comparisons
  • Support for statistical comparisons
  • Support for testing in-memory pipes without going to disk
• A ThermometerSink that would allow in memory fact checking
• Support for running full jobs in the same fact framework
• Support for re-running tests with different scalding modes
• Add the ability for Context to not depend on hdfs, via some sort of in memory representation for assertions.