Page Comparison

@Override
public void beforeSubmit(MapReduceContext context) throws Exception {
  ...
  // get the current state of the PartitionConsumer (user is currently responsible for storing state between runs)
  byte[] state = keyValueTable.read(STATE_KEY);
  PartitionConsumerState initialPartitionConsumerState;
  if (state == null) {
    initialPartitionConsumerState = PartitionConsumerState.FROM_BEGINNING;
  } else {
    initialPartitionConsumerState = PartitionConsumerState.fromBytes(state);
  }

  PartitionedFileSet results = context.getDataset("lines");
  PartitionConsumerResult partitionConsumerResult = results.consumePartitions(initialPartitionConsumerState);
  // keep track of the final state to store in onFinish
  finalPartitionConsumerState = partitionConsumerResult.getPartitionConsumerState();
 
  // set the dataset as the input, with the the specified partitions to process
  Map<String, String> inputArgs = Maps.newHashMap();
  PartitionedFileSetArguments.addPartitions(inputArgs, partitionConsumerResult.getPartitionIterator());
  PartitionedFileSet input = context.getDataset("lines", inputArgs);
  context.setInput("lines", input);
  ...
}

@Override
public void onFinish(boolean succeeded, MapReduceContext context) throws Exception {
  if (succeeded) {
    ...
    keyValueTable.write(STATE_KEY, finalPartitionConsumerState.toBytes());
  }
  super.onFinish(succeeded, context);
}