TestStreaming.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:4k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.streaming;
  19. import junit.framework.TestCase;
  20. import java.io.*;
  21. /**
  22.  * This class tests hadoopStreaming in MapReduce local mode.
  23.  */
  24. public class TestStreaming extends TestCase
  25. {
  26.   // "map" command: grep -E (red|green|blue)
  27.   // reduce command: uniq
  28.   protected File INPUT_FILE = new File("input.txt");
  29.   protected File OUTPUT_DIR = new File("out");
  30.   protected String input = "roses.are.rednviolets.are.bluenbunnies.are.pinkn";
  31.   // map behaves like "/usr/bin/tr . \n"; (split words into lines)
  32.   protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\n"});
  33.   // reduce behave like /usr/bin/uniq. But also prepend lines with R.
  34.   // command-line combiner does not have any effect any more.
  35.   protected String reduce = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"R"});
  36.   protected String outputExpect = "RaretnRbluetnRbunniestnRpinktnRredtnRrosestnRvioletstn";
  37.   private StreamJob job;
  38.   public TestStreaming() throws IOException
  39.   {
  40.     UtilTest utilTest = new UtilTest(getClass().getName());
  41.     utilTest.checkUserDir();
  42.     utilTest.redirectIfAntJunit();
  43.   }
  44.   protected void createInput() throws IOException
  45.   {
  46.     DataOutputStream out = new DataOutputStream(
  47.                                                 new FileOutputStream(INPUT_FILE.getAbsoluteFile()));
  48.     out.write(input.getBytes("UTF-8"));
  49.     out.close();
  50.   }
  51.   protected String[] genArgs() {
  52.     return new String[] {
  53.       "-input", INPUT_FILE.getAbsolutePath(),
  54.       "-output", OUTPUT_DIR.getAbsolutePath(),
  55.       "-mapper", map,
  56.       "-reducer", reduce,
  57.       //"-verbose",
  58.       //"-jobconf", "stream.debug=set"
  59.       "-jobconf", "keep.failed.task.files=true",
  60.       "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
  61.     };
  62.   }
  63.   
  64.   public void testCommandLine() throws IOException
  65.   {
  66.     try {
  67.       try {
  68.         OUTPUT_DIR.getAbsoluteFile().delete();
  69.       } catch (Exception e) {
  70.       }
  71.       createInput();
  72.       boolean mayExit = false;
  73.       // During tests, the default Configuration will use a local mapred
  74.       // So don't specify -config or -cluster
  75.       job = new StreamJob(genArgs(), mayExit);      
  76.       job.go();
  77.       File outFile = new File(OUTPUT_DIR, "part-00000").getAbsoluteFile();
  78.       String output = StreamUtil.slurp(outFile);
  79.       outFile.delete();
  80.       System.err.println("outEx1=" + outputExpect);
  81.       System.err.println("  out1=" + output);
  82.       assertEquals(outputExpect, output);
  83.     } finally {
  84.       File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
  85.       INPUT_FILE.delete();
  86.       outFileCRC.delete();
  87.       OUTPUT_DIR.getAbsoluteFile().delete();
  88.     }
  89.   }
  90.   public static void main(String[]args) throws Exception
  91.   {
  92.     new TestStreaming().testCommandLine();
  93.   }
  94. }