Java 类com.mongodb.hadoop.util.MongoConfigUtil 实例源码

@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
protected void setup(Mapper.Context context) throws IOException, InterruptedException {
    super.setup(context);

    ConfigSections cfg = JobConfiguration.fromHadoopConfiguration(context.getConfiguration());
    MetadataConfig meta = cfg.getMetadata();
    bands = meta.getCutPoints();

    BSONObject obj = MongoConfigUtil.getFields(context.getConfiguration());
    if (obj != null) {
        fields = obj.keySet().toArray(new String[0]);
    } else {
        throw new IllegalArgumentException("Invalid configuration found. Aggregates must "
            + "specify a the hadoop.map.fields property.");
    }
}

/**
 * Test method for
 * {@link org.slc.sli.aggregation.mapreduce.io.MongoAggFormatter#getRecordWriter(org.apache.hadoop.mapreduce.TaskAttemptContext)}
 * .
 *
 * @throws Exception
 */
@Test
public void testGetRecordWriter() throws Exception {

    DBCollection mockCollection = Mockito.mock(DBCollection.class);
    UserGroupInformation ugi = Mockito.mock(UserGroupInformation.class);

    PowerMockito.mockStatic(UserGroupInformation.class);

    Mockito.when(UserGroupInformation.getCurrentUser()).thenReturn(ugi);

    TaskAttemptContext c = new MockTaskAttemptContext();
    Configuration config = c.getConfiguration();

    PowerMockito.mockStatic(MongoConfigUtil.class);
    Mockito.when(MongoConfigUtil.getOutputCollection(config)).thenReturn(mockCollection);

    MongoAggFormatter f = new MongoAggFormatter();
    assertTrue(f.getRecordWriter(new MockTaskAttemptContext()) instanceof MongoAggWriter);
}

public void expect(T expected) throws IOException {
    this.expected = expected;

    mockCollection = Mockito.mock(DBCollection.class);

    ugi = Mockito.mock(UserGroupInformation.class);

    PowerMockito.mockStatic(UserGroupInformation.class);
    Mockito.when(UserGroupInformation.getCurrentUser()).thenReturn(ugi);

    ctx = new MockTaskAttemptContext();
    config = ctx.getConfiguration();

    PowerMockito.mockStatic(MongoConfigUtil.class);
    Mockito.when(MongoConfigUtil.getOutputCollection(config)).thenReturn(mockCollection);

    key = new TenantAndIdEmittableKey("testTenant", "testId");
    key.setTenantId(new Text("Midgar"));
    key.setId(new Text("abcdefg01234567890"));
    writer = new MongoAggWriter(mockCollection, ctx);
}

private void setupJob() {
  _job.setInputFormatClass(MongoInputFormat.class);
  _job.setMapperClass(BulkImportMapper.class);
  _job.setMapOutputKeyClass(ImmutableBytesWritable.class);
  _job.setMapOutputValueClass(Put.class);

  MongoConfigUtil.setInputURI(getConfiguration(), _mongoURI);
  MongoConfigUtil.setReadSplitsFromSecondary(getConfiguration(), true);
}

@Override
public RecordWriter<EmittableKey, BSONWritable> getRecordWriter(
    org.apache.hadoop.mapreduce.TaskAttemptContext context) {
    Configuration config = context.getConfiguration();
    return new MongoAggWriter(MongoConfigUtil.getOutputCollection(config), context);
}