org.apache.hadoop.io.ObjectWritable#set ( )源码实例Demo

/**
 * Changes input into ObjectWritables.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Convert values to ObjectWritable
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Convert values to ObjectWritable
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(WritableUtils.clone(value, conf));
  output.collect(key, objWrite);
}
 

/**
 * Wraps all values in ObjectWritables.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Wraps values in ObjectWritable.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

@Test
public void testMapper() throws IOException {

  final GeoWaveInputKey inputKey = new GeoWaveInputKey();
  inputKey.setInternalAdapterId(adapterId);
  inputKey.setDataId(new ByteArray("abc".getBytes()));

  final ObjectWritable ow = new ObjectWritable();
  ow.set(
      new FeatureWritable(
          ftype,
          AnalyticFeature.createGeometryFeature(
              ftype,
              batchId,
              "123",
              "fred",
              grp1,
              20.30203,
              factory.createPoint(new Coordinate(02.33, 0.23)),
              new String[] {"extra1"},
              new double[] {0.022},
              1,
              1,
              0)));

  mapDriver.withInput(inputKey, ow);

  final List<Pair<Text, CountofDoubleWritable>> results = mapDriver.run();
  // output key has the dataID adjusted to contain the rank
  assertEquals(results.get(0).getFirst().toString(), grp1);
  // output value is the same as input value
  assertEquals(results.get(0).getSecond().getValue(), 0.0, 0.0001);
}
 

@Test
public void testMapperWithZeroRank() throws IOException {
  capturedObjects.clear();
  mapDriver.getConfiguration().setClass(
      GeoWaveConfiguratorBase.enumToConfKey(
          KSamplerMapReduce.class,
          SampleParameters.Sample.SAMPLE_RANK_FUNCTION),
      TestSamplingNoRankFunction.class,
      SamplingRankFunction.class);

  final GeoWaveInputKey inputKey = new GeoWaveInputKey();
  inputKey.setInternalAdapterId(internalAdapterId);
  inputKey.setDataId(new ByteArray("abc".getBytes()));

  final ObjectWritable ow = new ObjectWritable();
  ow.set(new TestObjectWritable(new TestObject(new Coordinate(25.4, 25.6), "abc")));

  final GeoWaveInputKey outputKey = new GeoWaveInputKey();
  outputKey.setInternalAdapterId(internalAdapterId);

  final ByteBuffer keyBuf = ByteBuffer.allocate(64);
  keyBuf.putDouble(0.0);
  keyBuf.putInt(3);
  keyBuf.put(inputKey.getDataId().getBytes());
  outputKey.setDataId(new ByteArray(keyBuf.array()));

  mapDriver.withInput(inputKey, ow);

  final List<Pair<GeoWaveInputKey, ObjectWritable>> results = mapDriver.run();

  assertEquals(0, results.size());

  // results from sample rank function to make sure it was provided the
  // correct object
  assertEquals(1, capturedObjects.size());
  assertEquals("abc", ((TestObject) capturedObjects.get(0)).id);
}
 

/**
 * Changes input into ObjectWritables.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Convert values to ObjectWritable
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Convert values to ObjectWritable
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(WritableUtils.clone(value, conf));
  output.collect(key, objWrite);
}
 

/**
 * Wraps all values in ObjectWritables.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

/**
 * Wraps values in ObjectWritable.
 */
public void map(Text key, Writable value,
  OutputCollector<Text, ObjectWritable> output, Reporter reporter)
  throws IOException {

  ObjectWritable objWrite = new ObjectWritable();
  objWrite.set(value);
  output.collect(key, objWrite);
}
 

public static Writable toWritable(Object o) throws IOException {
  if (o instanceof Long) {
    return new LongWritable(((Long) o).longValue());
  } else if (o instanceof String) {
    return new Text(o.toString());
  } else if (o instanceof byte[]) {
    return new BytesWritable((byte[]) o);
  } else if (o instanceof Integer) {
    return new IntWritable(((Integer) o).intValue());
  } else if (o instanceof Short) {
    return new ShortWritable(((Short) o).shortValue());
  } else if (o instanceof Byte) {
    return new ByteWritable(((Byte) o).byteValue());
  } else if (o instanceof Double) {
    return new DoubleWritable(((Double) o).doubleValue());
  } else if (o instanceof Float) {
    return new FloatWritable(((Float) o).floatValue());
  } else if (o instanceof Boolean) {
    return new BooleanWritable(((Boolean) o).booleanValue());
  } else if (o instanceof List) {
    Writable[] a = new Writable[((List) o).size()];
    for (int i = 0; i < a.length; i++) {
      a[i] = new ObjectWritable(toWritable(((List) o).get(i)));
    }
    return new ArrayWritable(ObjectWritable.class, a);
  } else if (o instanceof Map) {
    MapWritable map = new MapWritable();
    for (Entry<Object,Object> entry: ((Map<Object,Object>) o).entrySet()) {
      map.put(toWritable(entry.getKey()), toWritable(entry.getValue()));
    }
    return map;
  } else if (null == o) {
    return NullWritable.get();
  } else {
    ObjectWritable ow = new ObjectWritable();
    ow.set(o);
    return ow;
  }// else {
  //  throw new IOException("Unsupported type " + o.getClass());
  //}
}
 

@Test
public void testMapperWithMidRankedKey() throws IOException {

  capturedObjects.clear();
  mapDriver.getConfiguration().setClass(
      GeoWaveConfiguratorBase.enumToConfKey(
          KSamplerMapReduce.class,
          SampleParameters.Sample.SAMPLE_RANK_FUNCTION),
      TestSamplingMidRankFunction.class,
      SamplingRankFunction.class);

  final GeoWaveInputKey inputKey = new GeoWaveInputKey();
  inputKey.setInternalAdapterId(internalAdapterId);
  inputKey.setDataId(new ByteArray("abc".getBytes()));

  final ObjectWritable ow = new ObjectWritable();
  ow.set(new TestObjectWritable(new TestObject(new Coordinate(25.4, 25.6), "abc")));

  final GeoWaveInputKey outputKey = new GeoWaveInputKey();
  outputKey.setInternalAdapterId(internalAdapterId);

  final ByteBuffer keyBuf = ByteBuffer.allocate(64);
  keyBuf.putDouble(0.5);
  keyBuf.putInt(1);
  keyBuf.put("1".getBytes());
  keyBuf.putInt(3);
  keyBuf.put(inputKey.getDataId().getBytes());
  outputKey.setDataId(new ByteArray(keyBuf.array()));

  mapDriver.withInput(inputKey, ow);

  final List<Pair<GeoWaveInputKey, ObjectWritable>> results = mapDriver.run();
  // output key has the dataID adjusted to contain the rank
  assertEquals(results.get(0).getFirst(), outputKey);
  // output value is the same as input value
  assertEquals(results.get(0).getSecond().get(), ow.get());

  // results from sample rank function to make sure it was provided the
  // correct object
  assertEquals(1, capturedObjects.size());
  assertEquals("abc", ((TestObject) capturedObjects.get(0)).id);
}
 

@Test
public void testReducer() throws IOException {

  final ObjectWritable ow1 = new ObjectWritable();
  ow1.set(new TestObjectWritable(new TestObject(new Coordinate(25.4, 25.6), "abc")));

  final ObjectWritable ow2 = new ObjectWritable();
  ow2.set(new TestObjectWritable(new TestObject(new Coordinate(25.4, 25.6), "def")));

  final ObjectWritable ow3 = new ObjectWritable();
  ow3.set(new TestObjectWritable(new TestObject(new Coordinate(25.4, 25.6), "ghi")));

  final GeoWaveInputKey inputKey1 = new GeoWaveInputKey();
  inputKey1.setInternalAdapterId(internalAdapterId);

  ByteBuffer keyBuf = ByteBuffer.allocate(64);
  keyBuf.putDouble(0.5);
  keyBuf.putInt(3);
  keyBuf.put("111".getBytes());
  inputKey1.setDataId(new ByteArray(keyBuf.array()));

  keyBuf = ByteBuffer.allocate(64);
  final GeoWaveInputKey inputKey2 = new GeoWaveInputKey();
  inputKey2.setInternalAdapterId(internalAdapterId);
  keyBuf.putDouble(0.6);
  keyBuf.putInt(3);
  keyBuf.put("111".getBytes());
  inputKey2.setDataId(new ByteArray(keyBuf.array()));

  keyBuf = ByteBuffer.allocate(64);
  final GeoWaveInputKey inputKey3 = new GeoWaveInputKey();
  inputKey3.setInternalAdapterId(internalAdapterId);
  keyBuf.putDouble(0.7);
  keyBuf.putInt(3);
  keyBuf.put("111".getBytes());
  inputKey3.setDataId(new ByteArray(keyBuf.array()));

  reduceDriver.addInput(inputKey1, Arrays.asList(ow1));

  reduceDriver.addInput(inputKey2, Arrays.asList(ow2));

  reduceDriver.addInput(inputKey3, Arrays.asList(ow3));

  final List<Pair<GeoWaveOutputKey, TestObject>> results = reduceDriver.run();
  assertEquals(2, results.size());
  assertEquals(results.get(0).getFirst().getTypeName(), "altoids");
  assertEquals(results.get(1).getFirst().getTypeName(), "altoids");
  assertEquals("abc", results.get(0).getSecond().getName());
  assertEquals("def", results.get(1).getSecond().getName());
}