Hadoop Counters in Scalding

The idea is to access the cascading FlowProcess through a cascading operator. We use an anonymous function for the operator which does nothing except incrementing a specified hadoop counter.

def addCounter(pipe : Pipe, group : String, counter : String) = {

  pipe.each(() -> ('addCounter)) { fields =>
    new BaseOperation[Any](fields) with Function[Any] {

      def operate(flowProcess : FlowProcess[_], 
        functionCall : FunctionCall[Any]) {

          try {
            flowProcess.asInstanceOf[HadoopFlowProcess]
              .increment(group, counter, 1L)
            functionCall.getOutputCollector.add(new Tuple(new Array[Object](1) : _*))
          } catch {
            case cce: ClassCastException =>
            // HadoopFlowProcess is not available in local mode
          }
      }
    }
  }.discard('addCounter)
}

In the scalding job, the tuples passing through any part of the Flow can be counted using code like this:

.then { p =>  addCounter(p, "my_group", "my_counter") }

The counter will appear on the job details page of the job in the job tracker, and will count all the tuples that make it past the filter.

Adapted from here.

Update (2013/08/29): had to add a dummy field which I finally discard. Otherwise, no tuples were being emitted when the method was called, even though they were being counted.