使用动易模版制作网站seo标签优化
这篇文章主要介绍了Java执行hadoop的基本操作实例代码的相关资料,需要的朋友可以参考下
Java执行hadoop的基本操作实例代码
向HDFS上传本地文件
public static void uploadInputFile(String localFile) throws IOException{
Configuration conf = new Configuration();
String hdfsPath = "hdfs://localhost:9000/";
String hdfsInput = "hdfs://localhost:9000/user/hadoop/input";
FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
fs.copyFromLocalFile(new Path(localFile), new Path(hdfsInput));
fs.close();
System.out.println("已经上传文件到input文件夹啦");
}
将output文件下载到本地
public static void getOutput(String outputfile) throws IOException{
String remoteFile = "hdfs://localhost:9000/user/hadoop/output/part-r-00000";
Path path = new Path(remoteFile);
Configuration conf = new Configuration();
String hdfsPath = "hdfs://localhost:9000/";
FileSystem fs = FileSystem.get(URI.create(hdfsPath),conf);
fs.copyToLocalFile(path, new Path(outputfile));
System.out.println("已经将输出文件保留到本地文件");
fs.close();
}
删除hdfs中的文件
public static void deleteOutput() throws IOException{
Configuration conf = new Configuration();
String hdfsOutput = "hdfs://localhost:9000/user/hadoop/output";
String hdfsPath = "hdfs://localhost:9000/";
Path path = new Path(hdfsOutput);
FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
fs.deleteOnExit(path);
fs.close();
System.out.println("output文件已经删除");
}
执行mapReduce程序
创建Mapper类和Reducer类
public static class TokenizerMapper extends Mapper{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context) throws IOException, InterruptedException{
String line = value.toString();
line = line.replace("\\", "");
String regex = "性别：(.*?)";
Pattern pattern = Pattern.compile(regex);
Matcher matcher = pattern.matcher(line);
while(matcher.find()){
String term = matcher.group(1);
word.set(term);
context.write(word, one);
}
}
}
public static class IntSumReducer extends Reducer{
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException{
int sum = 0;
for(IntWritable val :values){
sum+= val.get();
}
result.set(sum);
context.write(key, result);
}
}
执行mapReduce程序
public static void runMapReduce(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if(otherArgs.length != 2){
System.err.println("Usage: wordcount");
System.exit(2);
}
Job job = new Job(conf, "word count");
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, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.out.println("mapReduce 执行完毕！");
System.exit(job.waitForCompletion(true)?0:1);
}
感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！