TestStreamAggregate.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. import java.util.*;
  22. import org.apache.hadoop.conf.Configuration;
  23. import org.apache.hadoop.fs.FileSystem;
  24. import org.apache.hadoop.fs.Path;
  25. /**
  26.  * This class tests hadoopStreaming in MapReduce local mode.
  27.  * It uses Hadoop Aggregate to count the numbers of word occurrences 
  28.  * in the input.
  29.  */
  30. public class TestStreamAggregate extends TestCase
  31. {
  32.   protected File INPUT_FILE = new File("stream_aggregate_input.txt");
  33.   protected File OUTPUT_DIR = new File("stream_aggregate_out");
  34.   protected String input = "roses are rednviolets are bluenbunnies are pinkn";
  35.   // map parses input lines and generates count entries for each word.
  36.   protected String map = StreamUtil.makeJavaCommand(StreamAggregate.class, new String[]{".", "\n"});
  37.   // Use the aggregate combine, reducei to aggregate the counts
  38.   protected String outputExpect = "aret3nbluet1nbunniest1npinkt1nredt1nrosest1nvioletst1n";
  39.   private StreamJob job;
  40.   public TestStreamAggregate() throws IOException
  41.   {
  42.     UtilTest utilTest = new UtilTest(getClass().getName());
  43.     utilTest.checkUserDir();
  44.     utilTest.redirectIfAntJunit();
  45.   }
  46.   protected void createInput() throws IOException
  47.   {
  48.     DataOutputStream out = new DataOutputStream(
  49.                                                 new FileOutputStream(INPUT_FILE.getAbsoluteFile()));
  50.     out.write(input.getBytes("UTF-8"));
  51.     out.close();
  52.   }
  53.   protected String[] genArgs() {
  54.     return new String[] {
  55.       "-input", INPUT_FILE.getAbsolutePath(),
  56.       "-output", OUTPUT_DIR.getAbsolutePath(),
  57.       "-mapper", map,
  58.       "-reducer", "aggregate",
  59.       //"-verbose",
  60.       //"-jobconf", "stream.debug=set"
  61.       "-jobconf", "keep.failed.task.files=true",
  62.       "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
  63.     };
  64.   }
  65.   
  66.   public void testCommandLine()
  67.   {
  68.     try {
  69.       try {
  70.         OUTPUT_DIR.getAbsoluteFile().delete();
  71.       } catch (Exception e) {
  72.       }
  73.       createInput();
  74.       boolean mayExit = false;
  75.       // During tests, the default Configuration will use a local mapred
  76.       // So don't specify -config or -cluster
  77.       job = new StreamJob(genArgs(), mayExit);      
  78.       job.go();
  79.       File outFile = new File(OUTPUT_DIR, "part-00000").getAbsoluteFile();
  80.       String output = StreamUtil.slurp(outFile);
  81.       outFile.delete();
  82.       System.err.println("outEx1=" + outputExpect);
  83.       System.err.println("  out1=" + output);
  84.       assertEquals(outputExpect, output);
  85.     } catch(Exception e) {
  86.       failTrace(e);
  87.     } finally {
  88.       File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
  89.       INPUT_FILE.delete();
  90.       outFileCRC.delete();
  91.       OUTPUT_DIR.getAbsoluteFile().delete();
  92.     }
  93.   }
  94.   private void failTrace(Exception e)
  95.   {
  96.     StringWriter sw = new StringWriter();
  97.     e.printStackTrace(new PrintWriter(sw));
  98.     fail(sw.toString());
  99.   }
  100.   public static void main(String[]args) throws Exception
  101.   {
  102.     new TestStreaming().testCommandLine();
  103.   }
  104. }