hadoop2集群安装和测试之demo测试

demo实现如下功能：对tomcat的访问日志，统计其客户端IP的访问次数。

新建Map/Reduce Project ：hadoop-demo

在src目录下创建log4j.properties文件，方便跟踪日志，内容如下： 

log4j.rootLogger=debug,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n
log4j.logger.com.codefutures=DEBUG

为了演示demo我们在hdfs的/in上传一些tomcat访问日志



我们通过程序统计给个IP的访问次数，创建类IPCount.java

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

public class IPCount {

public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
if(tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
}
}
}

public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}

public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(WordCount.class);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(Map.class);
conf.setReducerClass(Reduce.class);

conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(conf, new Path("hdfs://study-90:9000/in"));
FileOutputFormat.setOutputPath(conf, new Path("hdfs://study-90:9000/output1"));
JobClient.runJob(conf);
}
}

运行后，可以刷新DFS插件，查看结果



若运行出错：java.lang.UnsatisfiedLinkError:org.apache.hadoop.io.nativeio.

NativeIO$Windows.access0(Ljava/lang/String;I)Z

主要是权限问题，将源代码NativeIO.java放入工程的src目录下并创建同样的包名，然后修改源代码

public staticbooleanaccess(String path, AccessRight desiredAccess)

throws IOException {
return ture;
//  return access0(path,desiredAccess.accessRight());

}

找到源码将 return access0(path,desiredAccess.accessRight());注释掉，return true;重新编译运行即可。

其他解决办法，在IPCount类的main方法里添加System.setProperty("HADOOP_USER_NAME", "hadoop");其中hadoop为有权限登录hadoop系统的linux用户名。

出错问题解决可以参考：http://blog.csdn.net/u010911997/article/details/44099165