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

网格计算

开发平台:

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 with customized separator in MapReduce local mode.
  27.  */
  28. public class TestStreamingSeparator extends TestCase
  29. {
  30.   // "map" command: grep -E (red|green|blue)
  31.   // reduce command: uniq
  32.   protected File INPUT_FILE = new File("TestStreamingSeparator.input.txt");
  33.   protected File OUTPUT_DIR = new File("TestStreamingSeparator.out");
  34.   protected String input = "roses1are.rednviolets1are.bluenbunnies1are.pinkn";
  35.   // key.value.separator.in.input.line reads 1 as separator
  36.   // stream.map.input.field.separator uses 2 as separator
  37.   // map behaves like "/usr/bin/tr 2 3"; (translate 2 to 3)
  38.   protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{"2", "3"});
  39.   // stream.map.output.field.separator recognize 3 as separator
  40.   // stream.reduce.input.field.separator recognize 3 as separator
  41.   // reduce behaves like "/usr/bin/tr 3 4"; (translate 3 to 4)
  42.   protected String reduce = StreamUtil.makeJavaCommand(TrAppReduce.class, new String[]{"3", "4"});
  43.   // stream.reduce.output.field.separator recognize 4 as separator
  44.   // mapred.textoutputformat.separator outputs 5 as separator
  45.   protected String outputExpect = "bunnies5are.pinknroses5are.rednviolets5are.bluen";
  46.   private StreamJob job;
  47.   public TestStreamingSeparator() throws IOException
  48.   {
  49.     UtilTest utilTest = new UtilTest(getClass().getName());
  50.     utilTest.checkUserDir();
  51.     utilTest.redirectIfAntJunit();
  52.   }
  53.   protected void createInput() throws IOException
  54.   {
  55.     DataOutputStream out = new DataOutputStream(
  56.                                                 new FileOutputStream(INPUT_FILE.getAbsoluteFile()));
  57.     out.write(input.getBytes("UTF-8"));
  58.     out.close();
  59.   }
  60.   protected String[] genArgs() {
  61.     return new String[] {
  62.       "-input", INPUT_FILE.getAbsolutePath(),
  63.       "-output", OUTPUT_DIR.getAbsolutePath(),
  64.       "-mapper", map,
  65.       "-reducer", reduce,
  66.       //"-verbose",
  67.       //"-jobconf", "stream.debug=set"
  68.       "-jobconf", "keep.failed.task.files=true",
  69.       "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp"),
  70.       "-inputformat", "KeyValueTextInputFormat",
  71.       "-jobconf", "key.value.separator.in.input.line=1",
  72.       "-jobconf", "stream.map.input.field.separator=2",
  73.       "-jobconf", "stream.map.output.field.separator=3",
  74.       "-jobconf", "stream.reduce.input.field.separator=3",
  75.       "-jobconf", "stream.reduce.output.field.separator=4",
  76.       "-jobconf", "mapred.textoutputformat.separator=5",
  77.     };
  78.   }
  79.   
  80.   public void testCommandLine()
  81.   {
  82.     try {
  83.       try {
  84.         OUTPUT_DIR.getAbsoluteFile().delete();
  85.       } catch (Exception e) {
  86.       }
  87.       createInput();
  88.       boolean mayExit = false;
  89.       // During tests, the default Configuration will use a local mapred
  90.       // So don't specify -config or -cluster
  91.       job = new StreamJob(genArgs(), mayExit);      
  92.       job.go();
  93.       File outFile = new File(OUTPUT_DIR, "part-00000").getAbsoluteFile();
  94.       String output = StreamUtil.slurp(outFile);
  95.       outFile.delete();
  96.       System.err.println("outEx1=" + outputExpect);
  97.       System.err.println("  out1=" + output);
  98.       assertEquals(outputExpect, output);
  99.     } catch(Exception e) {
  100.       failTrace(e);
  101.     } finally {
  102.       File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
  103.       INPUT_FILE.delete();
  104.       outFileCRC.delete();
  105.       OUTPUT_DIR.getAbsoluteFile().delete();
  106.     }
  107.   }
  108.   private void failTrace(Exception e)
  109.   {
  110.     StringWriter sw = new StringWriter();
  111.     e.printStackTrace(new PrintWriter(sw));
  112.     fail(sw.toString());
  113.   }
  114.   public static void main(String[]args) throws Exception
  115.   {
  116.     new TestStreamingSeparator().testCommandLine();
  117.   }
  118. }