TestStreamAggregate.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.*;
- import java.util.*;
- import org.apache.hadoop.conf.Configuration;
- import org.apache.hadoop.fs.FileSystem;
- import org.apache.hadoop.fs.Path;
- /**
- * This class tests hadoopStreaming in MapReduce local mode.
- * It uses Hadoop Aggregate to count the numbers of word occurrences
- * in the input.
- */
- public class TestStreamAggregate extends TestCase
- {
- protected File INPUT_FILE = new File("stream_aggregate_input.txt");
- protected File OUTPUT_DIR = new File("stream_aggregate_out");
- protected String input = "roses are rednviolets are bluenbunnies are pinkn";
- // map parses input lines and generates count entries for each word.
- protected String map = StreamUtil.makeJavaCommand(StreamAggregate.class, new String[]{".", "\n"});
- // Use the aggregate combine, reducei to aggregate the counts
- protected String outputExpect = "aret3nbluet1nbunniest1npinkt1nredt1nrosest1nvioletst1n";
- private StreamJob job;
- public TestStreamAggregate() 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", "aggregate",
- //"-verbose",
- //"-jobconf", "stream.debug=set"
- "-jobconf", "keep.failed.task.files=true",
- "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
- };
- }
-
- public void testCommandLine()
- {
- 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);
- } catch(Exception e) {
- failTrace(e);
- } finally {
- File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
- INPUT_FILE.delete();
- outFileCRC.delete();
- OUTPUT_DIR.getAbsoluteFile().delete();
- }
- }
- private void failTrace(Exception e)
- {
- StringWriter sw = new StringWriter();
- e.printStackTrace(new PrintWriter(sw));
- fail(sw.toString());
- }
- public static void main(String[]args) throws Exception
- {
- new TestStreaming().testCommandLine();
- }
- }