| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.hadoop.mapred; |
| |
| import org.apache.hadoop.fs.*; |
| import org.apache.hadoop.io.*; |
| import org.apache.hadoop.mapred.lib.*; |
| import junit.framework.TestCase; |
| import java.io.*; |
| import java.util.*; |
| |
| /** |
| * TestMapOutputType checks whether the Map task handles type mismatch |
| * between mapper output and the type specified in |
| * JobConf.MapOutputKeyType and JobConf.MapOutputValueType. |
| */ |
| public class TestMapOutputType extends TestCase |
| { |
| JobConf conf = new JobConf(TestMapOutputType.class); |
| JobClient jc; |
| /** |
| * TextGen is a Mapper that generates a Text key-value pair. The |
| * type specified in conf will be anything but. |
| */ |
| |
| static class TextGen implements Mapper { |
| public void configure(JobConf job) { |
| } |
| |
| public void map(WritableComparable key, Writable val, OutputCollector out, |
| Reporter reporter) throws IOException { |
| key = new Text("Hello"); |
| val = new Text("World"); |
| |
| out.collect(key, val); |
| } |
| |
| public void close() { |
| } |
| } |
| |
| /** A do-nothing reducer class. We won't get this far, really. |
| * |
| */ |
| static class TextReduce implements Reducer { |
| |
| public void configure(JobConf job) { |
| } |
| |
| public void reduce(WritableComparable key, |
| Iterator values, |
| OutputCollector out, |
| Reporter reporter) throws IOException { |
| out.collect(new Text("Test"), new Text("Me")); |
| } |
| |
| public void close() { |
| } |
| } |
| |
| |
| public void configure() throws Exception { |
| Path testdir = new Path("build/test/test.mapred.spill"); |
| Path inDir = new Path(testdir, "in"); |
| Path outDir = new Path(testdir, "out"); |
| FileSystem fs = FileSystem.get(conf); |
| fs.delete(testdir); |
| conf.setInt("io.sort.mb", 1); |
| conf.setInputFormat(SequenceFileInputFormat.class); |
| conf.setInputPath(inDir); |
| conf.setOutputPath(outDir); |
| conf.setMapperClass(TextGen.class); |
| conf.setReducerClass(TextReduce.class); |
| conf.setOutputKeyClass(Text.class); |
| conf.setOutputValueClass(Text.class); |
| |
| conf.setOutputFormat(SequenceFileOutputFormat.class); |
| if (!fs.mkdirs(testdir)) { |
| throw new IOException("Mkdirs failed to create " + testdir.toString()); |
| } |
| if (!fs.mkdirs(inDir)) { |
| throw new IOException("Mkdirs failed to create " + inDir.toString()); |
| } |
| Path inFile = new Path(inDir, "part0"); |
| SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, inFile, |
| Text.class, Text.class); |
| writer.append(new Text("rec: 1"), new Text("Hello")); |
| writer.close(); |
| |
| jc = new JobClient(conf); |
| } |
| |
| public void testKeyMismatch() throws Exception { |
| configure(); |
| |
| // Set bad MapOutputKeyClass and MapOutputValueClass |
| conf.setMapOutputKeyClass(IntWritable.class); |
| conf.setMapOutputValueClass(IntWritable.class); |
| |
| RunningJob r_job = jc.submitJob(conf); |
| while (!r_job.isComplete()) { |
| Thread.sleep(1000); |
| } |
| |
| if (r_job.isSuccessful()) { |
| fail("Oops! The job was supposed to break due to an exception"); |
| } |
| } |
| |
| public void testValueMismatch() throws Exception { |
| configure(); |
| |
| // Set good MapOutputKeyClass, bad MapOutputValueClass |
| conf.setMapOutputKeyClass(Text.class); |
| conf.setMapOutputValueClass(IntWritable.class); |
| |
| RunningJob r_job = jc.submitJob(conf); |
| while (!r_job.isComplete()) { |
| Thread.sleep(1000); |
| } |
| |
| if (r_job.isSuccessful()) { |
| fail("Oops! The job was supposed to break due to an exception"); |
| } |
| } |
| |
| public void testNoMismatch() throws Exception{ |
| configure(); |
| |
| // Set good MapOutputKeyClass and MapOutputValueClass |
| conf.setMapOutputKeyClass(Text.class); |
| conf.setMapOutputValueClass(Text.class); |
| |
| RunningJob r_job = jc.submitJob(conf); |
| while (!r_job.isComplete()) { |
| Thread.sleep(1000); |
| } |
| |
| if (!r_job.isSuccessful()) { |
| fail("Oops! The job broke due to an unexpected error"); |
| } |
| } |
| } |