private static List<String> readSplit(FixedLengthInputFormat format, InputSplit split, Job job) throws Exception { List<String> result = new ArrayList<String>(); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); RecordReader<LongWritable, BytesWritable> reader = format.createRecordReader(split, context); MapContext<LongWritable, BytesWritable, LongWritable, BytesWritable> mcontext = new MapContextImpl<LongWritable, BytesWritable, LongWritable, BytesWritable>(job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); LongWritable key; BytesWritable value; while (reader.nextKeyValue()) { key = reader.getCurrentKey(); value = reader.getCurrentValue(); result.add(new String(value.getBytes(), 0, value.getLength())); } reader.close(); return result; }
private static List<Text> readSplit(KeyValueTextInputFormat format, InputSplit split, Job job) throws IOException, InterruptedException { List<Text> result = new ArrayList<Text>(); Configuration conf = job.getConfiguration(); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(conf); RecordReader<Text, Text> reader = format.createRecordReader(split, MapReduceTestUtil.createDummyMapTaskAttemptContext(conf)); MapContext<Text, Text, Text, Text> mcontext = new MapContextImpl<Text, Text, Text, Text>(conf, context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); while (reader.nextKeyValue()) { result.add(new Text(reader.getCurrentValue())); } reader.close(); return result; }
private static List<Text> readSplit(InputFormat<LongWritable,Text> format, InputSplit split, Job job) throws IOException, InterruptedException { List<Text> result = new ArrayList<Text>(); Configuration conf = job.getConfiguration(); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(conf); RecordReader<LongWritable, Text> reader = format.createRecordReader(split, MapReduceTestUtil.createDummyMapTaskAttemptContext(conf)); MapContext<LongWritable,Text,LongWritable,Text> mcontext = new MapContextImpl<LongWritable,Text,LongWritable,Text>(conf, context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); while (reader.nextKeyValue()) { result.add(new Text(reader.getCurrentValue())); } return result; }
@SuppressWarnings({"rawtypes", "unchecked"}) @Test (timeout=10000) public void testLoadMapper() throws Exception { Configuration conf = new Configuration(); conf.setInt(JobContext.NUM_REDUCES, 2); CompressionEmulationUtil.setCompressionEmulationEnabled(conf, true); conf.setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, true); TaskAttemptID taskId = new TaskAttemptID(); RecordReader<NullWritable, GridmixRecord> reader = new FakeRecordReader(); LoadRecordGkGrWriter writer = new LoadRecordGkGrWriter(); OutputCommitter committer = new CustomOutputCommitter(); StatusReporter reporter = new TaskAttemptContextImpl.DummyReporter(); LoadSplit split = getLoadSplit(); MapContext<NullWritable, GridmixRecord, GridmixKey, GridmixRecord> mapContext = new MapContextImpl<NullWritable, GridmixRecord, GridmixKey, GridmixRecord>( conf, taskId, reader, writer, committer, reporter, split); // context Context ctx = new WrappedMapper<NullWritable, GridmixRecord, GridmixKey, GridmixRecord>() .getMapContext(mapContext); reader.initialize(split, ctx); ctx.getConfiguration().setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, true); CompressionEmulationUtil.setCompressionEmulationEnabled( ctx.getConfiguration(), true); LoadJob.LoadMapper mapper = new LoadJob.LoadMapper(); // setup, map, clean mapper.run(ctx); Map<GridmixKey, GridmixRecord> data = writer.getData(); // check result assertEquals(2, data.size()); }
private static List<Text> readSplit(KeyValueTextInputFormat format, InputSplit split, Job job) throws IOException, InterruptedException { List<Text> result = new ArrayList<Text>(); Configuration conf = job.getConfiguration(); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(conf); RecordReader<Text, Text> reader = format.createRecordReader(split, MapReduceTestUtil.createDummyMapTaskAttemptContext(conf)); MapContext<Text, Text, Text, Text> mcontext = new MapContextImpl<Text, Text, Text, Text>(conf, context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); while (reader.nextKeyValue()) { result.add(new Text(reader.getCurrentValue())); } return result; }
/** * Utility to generate dummy Mapper#Context for use in Giraph internals. * This is the "key hack" to inject MapReduce-related data structures * containing YARN cluster metadata (and our GiraphConf from the AppMaster) * into our Giraph BSP task code. * @param tid the TaskAttemptID to construct this Mapper#Context from. * @return sort of a Mapper#Context if you squint just right. */ private Context buildProxyMapperContext(final TaskAttemptID tid) { MapContext mc = new MapContextImpl<Object, Object, Object, Object>( conf, // our Configuration, populated back at the GiraphYarnClient. tid, // our TaskAttemptId, generated w/YARN app, container, attempt IDs null, // RecordReader here will never be used by Giraph null, // RecordWriter here will never be used by Giraph null, // OutputCommitter here will never be used by Giraph new TaskAttemptContextImpl.DummyReporter() { // goes in task logs for now @Override public void setStatus(String msg) { LOG.info("[STATUS: task-" + bspTaskId + "] " + msg); } }, null); // Input split setting here will never be used by Giraph // now, we wrap our MapContext ref so we can produce a Mapper#Context WrappedMapper<Object, Object, Object, Object> wrappedMapper = new WrappedMapper<Object, Object, Object, Object>(); return wrappedMapper.getMapContext(mc); }
public StubContext(Configuration conf, RecordReader<Text, CopyListingFileStatus> reader, int taskId) throws IOException, InterruptedException { WrappedMapper<Text, CopyListingFileStatus, Text, Text> wrappedMapper = new WrappedMapper<>(); MapContextImpl<Text, CopyListingFileStatus, Text, Text> contextImpl = new MapContextImpl<>(conf, getTaskAttemptID(taskId), reader, writer, null, reporter, null); this.reader = reader; mapperContext = wrappedMapper.getMapContext(contextImpl); }
void checkFormat(Job job, int expectedN, int lastN) throws IOException, InterruptedException { NLineInputFormat format = new NLineInputFormat(); List<InputSplit> splits = format.getSplits(job); int count = 0; for (int i = 0; i < splits.size(); i++) { assertEquals("There are no split locations", 0, splits.get(i).getLocations().length); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); RecordReader<LongWritable, Text> reader = format.createRecordReader( splits.get(i), context); Class<?> clazz = reader.getClass(); assertEquals("reader class is LineRecordReader.", LineRecordReader.class, clazz); MapContext<LongWritable, Text, LongWritable, Text> mcontext = new MapContextImpl<LongWritable, Text, LongWritable, Text>( job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), splits.get(i)); reader.initialize(splits.get(i), mcontext); try { count = 0; while (reader.nextKeyValue()) { count++; } } finally { reader.close(); } if ( i == splits.size() - 1) { assertEquals("number of lines in split(" + i + ") is wrong" , lastN, count); } else { assertEquals("number of lines in split(" + i + ") is wrong" , expectedN, count); } } }
private int countRecords(int numSplits) throws IOException, InterruptedException { InputFormat<Text, BytesWritable> format = new SequenceFileInputFilter<Text, BytesWritable>(); if (numSplits == 0) { numSplits = random.nextInt(MAX_LENGTH / (SequenceFile.SYNC_INTERVAL / 20)) + 1; } FileInputFormat.setMaxInputSplitSize(job, fs.getFileStatus(inFile).getLen() / numSplits); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); // check each split int count = 0; for (InputSplit split : format.getSplits(job)) { RecordReader<Text, BytesWritable> reader = format.createRecordReader(split, context); MapContext<Text, BytesWritable, Text, BytesWritable> mcontext = new MapContextImpl<Text, BytesWritable, Text, BytesWritable>( job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); try { while (reader.nextKeyValue()) { LOG.info("Accept record " + reader.getCurrentKey().toString()); count++; } } finally { reader.close(); } } return count; }
/** * Test with no record length set. */ @Test (timeout=5000) public void testNoRecordLength() throws Exception { localFs.delete(workDir, true); Path file = new Path(workDir, new String("testFormat.txt")); createFile(file, null, 10, 10); // Create the job and do not set fixed record length Job job = Job.getInstance(defaultConf); FileInputFormat.setInputPaths(job, workDir); FixedLengthInputFormat format = new FixedLengthInputFormat(); List<InputSplit> splits = format.getSplits(job); boolean exceptionThrown = false; for (InputSplit split : splits) { try { TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); RecordReader<LongWritable, BytesWritable> reader = format.createRecordReader(split, context); MapContext<LongWritable, BytesWritable, LongWritable, BytesWritable> mcontext = new MapContextImpl<LongWritable, BytesWritable, LongWritable, BytesWritable>(job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); } catch(IOException ioe) { exceptionThrown = true; LOG.info("Exception message:" + ioe.getMessage()); } } assertTrue("Exception for not setting record length:", exceptionThrown); }
/** * Test with record length set to 0 */ @Test (timeout=5000) public void testZeroRecordLength() throws Exception { localFs.delete(workDir, true); Path file = new Path(workDir, new String("testFormat.txt")); createFile(file, null, 10, 10); Job job = Job.getInstance(defaultConf); // Set the fixed length record length config property FixedLengthInputFormat format = new FixedLengthInputFormat(); format.setRecordLength(job.getConfiguration(), 0); FileInputFormat.setInputPaths(job, workDir); List<InputSplit> splits = format.getSplits(job); boolean exceptionThrown = false; for (InputSplit split : splits) { try { TaskAttemptContext context = MapReduceTestUtil.createDummyMapTaskAttemptContext( job.getConfiguration()); RecordReader<LongWritable, BytesWritable> reader = format.createRecordReader(split, context); MapContext<LongWritable, BytesWritable, LongWritable, BytesWritable> mcontext = new MapContextImpl<LongWritable, BytesWritable, LongWritable, BytesWritable>(job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); } catch(IOException ioe) { exceptionThrown = true; LOG.info("Exception message:" + ioe.getMessage()); } } assertTrue("Exception for zero record length:", exceptionThrown); }
/** * Test with record length set to a negative value */ @Test (timeout=5000) public void testNegativeRecordLength() throws Exception { localFs.delete(workDir, true); Path file = new Path(workDir, new String("testFormat.txt")); createFile(file, null, 10, 10); // Set the fixed length record length config property Job job = Job.getInstance(defaultConf); FixedLengthInputFormat format = new FixedLengthInputFormat(); format.setRecordLength(job.getConfiguration(), -10); FileInputFormat.setInputPaths(job, workDir); List<InputSplit> splits = format.getSplits(job); boolean exceptionThrown = false; for (InputSplit split : splits) { try { TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); RecordReader<LongWritable, BytesWritable> reader = format.createRecordReader(split, context); MapContext<LongWritable, BytesWritable, LongWritable, BytesWritable> mcontext = new MapContextImpl<LongWritable, BytesWritable, LongWritable, BytesWritable>(job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); reader.initialize(split, mcontext); } catch(IOException ioe) { exceptionThrown = true; LOG.info("Exception message:" + ioe.getMessage()); } } assertTrue("Exception for negative record length:", exceptionThrown); }
private static List<String> readSplit(FixedLengthInputFormat format, InputSplit split, Job job) throws Exception { List<String> result = new ArrayList<String>(); TaskAttemptContext context = MapReduceTestUtil. createDummyMapTaskAttemptContext(job.getConfiguration()); RecordReader<LongWritable, BytesWritable> reader = format.createRecordReader(split, context); MapContext<LongWritable, BytesWritable, LongWritable, BytesWritable> mcontext = new MapContextImpl<LongWritable, BytesWritable, LongWritable, BytesWritable>(job.getConfiguration(), context.getTaskAttemptID(), reader, null, null, MapReduceTestUtil.createDummyReporter(), split); LongWritable key; BytesWritable value; try { reader.initialize(split, mcontext); while (reader.nextKeyValue()) { key = reader.getCurrentKey(); value = reader.getCurrentValue(); result.add(new String(value.getBytes(), 0, value.getLength())); } } finally { reader.close(); } return result; }
MapRunner(Context context) throws IOException, InterruptedException { mapper = ReflectionUtils.newInstance(mapClass, context.getConfiguration()); MapContext<K1, V1, K2, V2> mapContext = new MapContextImpl<K1, V1, K2, V2>(outer.getConfiguration(), outer.getTaskAttemptID(), reader, new SubMapRecordWriter(), context.getOutputCommitter(), new SubMapStatusReporter(), outer.getInputSplit()); subcontext = new WrappedMapper<K1, V1, K2, V2>().getMapContext(mapContext); reader.initialize(context.getInputSplit(), context); }
@Test public void testCloneMapContext() throws Exception { TaskID taskId = new TaskID(jobId, TaskType.MAP, 0); TaskAttemptID taskAttemptid = new TaskAttemptID(taskId, 0); MapContext<IntWritable, IntWritable, IntWritable, IntWritable> mapContext = new MapContextImpl<IntWritable, IntWritable, IntWritable, IntWritable>( conf, taskAttemptid, null, null, null, null, null); Mapper<IntWritable, IntWritable, IntWritable, IntWritable>.Context mapperContext = new WrappedMapper<IntWritable, IntWritable, IntWritable, IntWritable>().getMapContext( mapContext); ContextFactory.cloneMapContext(mapperContext, conf, null, null); }
public StubContext(Configuration conf, RecordReader<Text, CopyListingFileStatus> reader, int taskId) throws IOException, InterruptedException { WrappedMapper<Text, CopyListingFileStatus, Text, Text> wrappedMapper = new WrappedMapper<Text, CopyListingFileStatus, Text, Text>(); MapContextImpl<Text, CopyListingFileStatus, Text, Text> contextImpl = new MapContextImpl<Text, CopyListingFileStatus, Text, Text>(conf, getTaskAttemptID(taskId), reader, writer, null, reporter, null); this.reader = reader; this.mapperContext = wrappedMapper.getMapContext(contextImpl); }
