gistfile1.txt

public class FirehoseFn extends BaseOperation implements Function {

    public FirehoseFn() {
	super(2, new Fields("daterange", "duration"));
    }

    public void operate( FlowProcess flowProcess, FunctionCall functionCall ) {
	TupleEntry arguments = functionCall.getArguments();
	Tuple result = new Tuple();

	String datemillisBytes = (String) arguments.get(0);
	Long datemillis = Bytes.toLong(datemillisBytes.getBytes());
	Long range = (datemillis / 5000) * 5000;
	result.add(range);

	String durationBytes = (String) arguments.get(1);
	BigDecimal duration = Bytes.toBigDecimal(durationBytes.getBytes());
	result.add(duration.doubleValue());

	functionCall.getOutputCollector().add(result);
    }
}