Random number in Cascalog

gistfile1.txt

import cascading.flow.FlowProcess;
import cascading.flow.hadoop.HadoopFlowProcess;
import cascading.operation.FunctionCall;
import cascading.operation.OperationCall;
import cascading.tuple.Tuple;
import java.util.Random;
import cascalog.CascalogFunction;

public class RandInt extends CascalogFunction {
  long _seed;
  Random _rand;
  Integer _max;

  public RandInt() {
    _seed = new Random().nextLong();
    _max = null;
  }

  public RandInt(int max) {
    this();
    _max = max;
  }

  public void prepare(FlowProcess flowProcess, OperationCall operationCall) {
    _rand = new Random(_seed + ((HadoopFlowProcess) flowProcess).getCurrentTaskNum());
  }

  public void operate(FlowProcess flow_process, FunctionCall fn_call) {
      int rand;
      if(_max==null) rand = _rand.nextInt();
      else rand = _rand.nextInt(_max);
      fn_call.getOutputCollector().add(new Tuple(rand));
  }
}

I think this is related to your Randomness post at http://blog.rapleaf.com/dev/2009/08/14/using-random-numbers-in-mapreduce-is-dangerous/

The post makes sense, as you suggest distributing the seed to the tasks through the JobConf. But I don't see that happening in this gist. Seems to me that each instance of RandInt (one per task or tasktracker?) will get it's own seed, so if TaskTracker is lost, a re-execution would get a new seed.

Or is there some magic I'm missing, here?

The seed is chosen before the job starts (in the constructor). Cascading then serializes the RandInt instance for use in the tasks. The tasks then call "prepare" before they start processing records. If a task fails, it reuses the same RandInt instance with the same seed.

Oh.. cool. I didn't realize Cascading serialized the instance.