TestStreaming.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:4k
- /**
- * 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.streaming;
- import junit.framework.TestCase;
- import java.io.*;
- /**
- * This class tests hadoopStreaming in MapReduce local mode.
- */
- public class TestStreaming extends TestCase
- {
- // "map" command: grep -E (red|green|blue)
- // reduce command: uniq
- protected File INPUT_FILE = new File("input.txt");
- protected File OUTPUT_DIR = new File("out");
- protected String input = "roses.are.rednviolets.are.bluenbunnies.are.pinkn";
- // map behaves like "/usr/bin/tr . \n"; (split words into lines)
- protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\n"});
- // reduce behave like /usr/bin/uniq. But also prepend lines with R.
- // command-line combiner does not have any effect any more.
- protected String reduce = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"R"});
- protected String outputExpect = "RaretnRbluetnRbunniestnRpinktnRredtnRrosestnRvioletstn";
- private StreamJob job;
- public TestStreaming() throws IOException
- {
- UtilTest utilTest = new UtilTest(getClass().getName());
- utilTest.checkUserDir();
- utilTest.redirectIfAntJunit();
- }
- protected void createInput() throws IOException
- {
- DataOutputStream out = new DataOutputStream(
- new FileOutputStream(INPUT_FILE.getAbsoluteFile()));
- out.write(input.getBytes("UTF-8"));
- out.close();
- }
- protected String[] genArgs() {
- return new String[] {
- "-input", INPUT_FILE.getAbsolutePath(),
- "-output", OUTPUT_DIR.getAbsolutePath(),
- "-mapper", map,
- "-reducer", reduce,
- //"-verbose",
- //"-jobconf", "stream.debug=set"
- "-jobconf", "keep.failed.task.files=true",
- "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
- };
- }
-
- public void testCommandLine() throws IOException
- {
- try {
- try {
- OUTPUT_DIR.getAbsoluteFile().delete();
- } catch (Exception e) {
- }
- createInput();
- boolean mayExit = false;
- // During tests, the default Configuration will use a local mapred
- // So don't specify -config or -cluster
- job = new StreamJob(genArgs(), mayExit);
- job.go();
- File outFile = new File(OUTPUT_DIR, "part-00000").getAbsoluteFile();
- String output = StreamUtil.slurp(outFile);
- outFile.delete();
- System.err.println("outEx1=" + outputExpect);
- System.err.println(" out1=" + output);
- assertEquals(outputExpect, output);
- } finally {
- File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
- INPUT_FILE.delete();
- outFileCRC.delete();
- OUTPUT_DIR.getAbsoluteFile().delete();
- }
- }
- public static void main(String[]args) throws Exception
- {
- new TestStreaming().testCommandLine();
- }
- }