DataFu's Hourglass: Example 1

example1_1.json

{
  "type" : "record", "name" : "ExampleEvent", 
  "namespace" : "datafu.hourglass.test",
  "fields" : [ {
    "name" : "id",
    "type" : "long",
    "doc" : "ID"
  } ]
}

example1_10.java

job.setCombinerAccumulator(job.getReducerAccumulator());
job.setUseCombiner(true);

example1_11.java

job.run();

example1_12.json

{"key": {"member_id": 1}, "value": {"count": 5}}
{"key": {"member_id": 2}, "value": {"count": 3}}
{"key": {"member_id": 3}, "value": {"count": 3}}

example1_13.txt

2013/03/17:
{"id": 1}, {"id": 1}, {"id": 2}, {"id": 2}, {"id": 2},
{"id": 3}, {"id": 3}

example1_14.json

{"key": {"member_id": 1}, "value": {"count": 7}}
{"key": {"member_id": 2}, "value": {"count": 6}}
{"key": {"member_id": 3}, "value": {"count": 5}}

example1_2.txt

2013/03/15:
{"id": 1}, {"id": 1}, {"id": 1}, {"id": 2}, {"id": 3}, {"id": 3}

2013/03/16:
{"id": 1}, {"id": 1}, {"id": 2}, {"id": 2}, {"id": 3}, 

example1_3.java

PartitionCollapsingIncrementalJob job = 
    new PartitionCollapsingIncrementalJob(Example.class);

example1_4.java

final String namespace = "com.example";

final Schema keySchema = 
  Schema.createRecord("Key",null,namespace,false);

keySchema.setFields(Arrays.asList(
  new Field("member_id",Schema.create(Type.LONG),null,null)));

final String keySchemaString = keySchema.toString(true);

final Schema valueSchema = 
  Schema.createRecord("Value",null,namespace,false);

valueSchema.setFields(Arrays.asList(
  new Field("count",Schema.create(Type.INT),null,null)));

final String valueSchemaString = valueSchema.toString(true);

example1_5.json

{
  "type" : "record", "name" : "Key", "namespace" : "com.example",
  "fields" : [ {
    "name" : "member_id",
    "type" : "long"
  } ]
}

{
  "type" : "record", "name" : "Value", "namespace" : "com.example",
  "fields" : [ {
    "name" : "count",
    "type" : "int"
  } ]
}

example1_6.java

job.setKeySchema(keySchema);
job.setIntermediateValueSchema(valueSchema);
job.setOutputValueSchema(valueSchema);

example1_7.java

job.setInputPaths(Arrays.asList(new Path("/data/event")));
job.setOutputPath(new Path("/output"));
job.setReusePreviousOutput(true);

example1_8.java

job.setMapper(new Mapper<GenericRecord,GenericRecord,GenericRecord>()
{
  private transient Schema kSchema;
  private transient Schema vSchema;
  
  @Override
  public void map(
    GenericRecord input,
    KeyValueCollector<GenericRecord, GenericRecord> collector) 
  throws IOException, InterruptedException 
  {
    if (kSchema == null) 
      kSchema = new Schema.Parser().parse(keySchemaString);

    if (vSchema == null) 
      vSchema = new Schema.Parser().parse(valueSchemaString);

    GenericRecord key = new GenericData.Record(kSchema);
    key.put("member_id", input.get("id"));

    GenericRecord value = new GenericData.Record(vSchema);
    value.put("count", 1);

    collector.collect(key,value);
  }      
});

example1_9.java

job.setReducerAccumulator(new Accumulator<GenericRecord,GenericRecord>() 
{
  private transient int count;
  private transient Schema vSchema;
  
  @Override
  public void accumulate(GenericRecord value) {
    this.count += (Integer)value.get("count");
  }

  @Override
  public GenericRecord getFinal() {
    if (vSchema == null) 
      vSchema = new Schema.Parser().parse(valueSchemaString);

    GenericRecord output = new GenericData.Record(vSchema);
    output.put("count", count);

    return output;
  }

  @Override
  public void cleanup() {
    this.count = 0;
  }      
});