(final String tmpPrefix, final Configuration mrConfig, final String formulaStr, final Path pathIn)
| 92 | |
| 93 | |
| 94 | private static WritableVariableList ruDefStep(final String tmpPrefix, final Configuration mrConfig, |
| 95 | final String formulaStr, final Path pathIn) throws IOException, InterruptedException, ClassNotFoundException { |
| 96 | // run the job |
| 97 | final Path pathOut = new Path(tmpPrefix + "_MPRedDefOut"); |
| 98 | // write side-channel information |
| 99 | mrConfig.set(FORMULAFIELD, formulaStr); |
| 100 | final Job job = WritableUtils.newJob(mrConfig); |
| 101 | job.setJarByClass(MapRedScan.class); |
| 102 | job.setJobName("MapRevDefStep"); |
| 103 | FileInputFormat.setMaxInputSplitSize(job,4*1024*1024L); |
| 104 | FileInputFormat.setMinInputSplitSize(job,4*1024L); |
| 105 | FileInputFormat.addInputPath(job, pathIn); |
| 106 | FileOutputFormat.setOutputPath(job, pathOut); |
| 107 | job.setMapperClass(DefMapper.class); |
| 108 | job.setCombinerClass(DefReducer.class); |
| 109 | job.setReducerClass(DefReducer.class); |
| 110 | job.setOutputKeyClass(LongWritable.class); |
| 111 | job.setOutputValueClass(WritableVariableList.class); |
| 112 | job.setInputFormatClass(TextInputFormat.class); |
| 113 | job.setOutputFormatClass(TextOutputFormat.class); |
| 114 | job.setNumReduceTasks(1); |
| 115 | WritableVariableList r = null; |
| 116 | if(job.waitForCompletion(false)) { |
| 117 | // collect results |
| 118 | final FSDataInputStream fdi = pathOut.getFileSystem(mrConfig).open(new Path(pathOut,"part-r-00000")); |
| 119 | final BufferedReader d = new BufferedReader(new InputStreamReader(fdi)); |
| 120 | final String line = d.readLine(); |
| 121 | d.close(); |
| 122 | final String[] flds = line.split("\t",2); |
| 123 | r = WritableVariableList.fromString(flds[1]); |
| 124 | } |
| 125 | // clean up |
| 126 | pathOut.getFileSystem(mrConfig).delete(pathOut,true); |
| 127 | return r; |
| 128 | } |
| 129 | |
| 130 | |
| 131 |
no test coverage detected