Java执行hadoop的基本操作实例代码

2024-02-25 02:27:03 274

Java执行hadoop的基本操作实例代码

向HDFS上传本地文件

publicstaticvoiduploadInputFile(StringlocalFile)throwsIOException{
Configurationconf=newConfiguration();
StringhdfsPath="hdfs://localhost:9000/";
StringhdfsInput="hdfs://localhost:9000/user/hadoop/input";
FileSystemfs=FileSystem.get(URI.create(hdfsPath),conf);
fs.copyFromLocalFile(newPath(localFile),newPath(hdfsInput));
fs.close();
System.out.println("已经上传文件到input文件夹啦");
}

将output文件下载到本地

publicstaticvoidgetOutput(Stringoutputfile)throwsIOException{
StringremoteFile="hdfs://localhost:9000/user/hadoop/output/part-r-00000";
Pathpath=newPath(remoteFile);
Configurationconf=newConfiguration();
StringhdfsPath="hdfs://localhost:9000/";
FileSystemfs=FileSystem.get(URI.create(hdfsPath),conf);
fs.copyToLocalFile(path,newPath(outputfile));
System.out.println("已经将输出文件保留到本地文件");
fs.close();
}

删除hdfs中的文件

publicstaticvoiddeleteOutput()throwsIOException{
Configurationconf=newConfiguration();
StringhdfsOutput="hdfs://localhost:9000/user/hadoop/output";
StringhdfsPath="hdfs://localhost:9000/";
Pathpath=newPath(hdfsOutput);
FileSystemfs=FileSystem.get(URI.create(hdfsPath),conf);
fs.deleteOnExit(path);
fs.close();
System.out.println("output文件已经删除");
}

执行mapReduce程序

创建Mapper类和Reducer类

publicstaticclassTokenizerMapperextendsMapper{

privatefinalstaticIntWritableone=newIntWritable(1);
privateTextword=newText();

publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{
Stringline=value.toString();
line=line.replace("\\","");
Stringregex="性别：(.*?)";
Patternpattern=Pattern.compile(regex);
Matchermatcher=pattern.matcher(line);
while(matcher.find()){
Stringterm=matcher.group(1);
word.set(term);
context.write(word,one);
}
}
}

publicstaticclassIntSumReducerextendsReducer{

privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterablevalues,Contextcontext)throwsIOException,InterruptedException{
intsum=0;
for(IntWritableval:values){
sum+=val.get();
}
result.set(sum);
context.write(key,result);
}
}

执行mapReduce程序

publicstaticvoidrunMapReduce(String[]args)throwsException{
Configurationconf=newConfiguration();
String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();
if(otherArgs.length!=2){
System.err.println("Usage:wordcount");
System.exit(2);
}
Jobjob=newJob(conf,"wordcount");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job,newPath(otherArgs[0]));
FileOutputFormat.setOutputPath(job,newPath(otherArgs[1]));
System.out.println("mapReduce执行完毕！");
System.exit(job.waitForCompletion(true)?0:1);

}

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

Java执行hadoop的基本操作实例代码

热门推荐

随机推荐