任务需求
统计HDFS上文件的
wordcount
,并将统计结果输出到HDFS
功能拆解
- 读取HDFS文件
- 业务处理(词频统计)
- 缓存处理结果
- 将结果输出到HDFS
数据准备
- 事先往HDFS上传需要进行词频统计的文件
word.txt、word2.txt(可以是多个)...
- 假设目录是
/user/hadoop/input/...
框架搭建
先把具体的功能
框架搭建
出来,再进行细节方面
的编写。
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
public class HDFSWordCountDemo{
public static void main(String[] args) throws Exception{
// 1.读取HDFS文件
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "hadoop");
// 使用Java API取出HDFS指定目录下所有要进行词频统计的单词文件,false表示不需要递归
RemoteIterator<LocatedFileStatus> files = fs.listFiles(new Path("input"), false);
// 用于循环取出多个单词文本
while (files.hasNext()) {
LocatedFileStatus file = files.next();
FSDataInputStream in = fs.open(file.getPath());
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line = null; // readLine每次读取一行
// 用于循环取出每个文本的每行内容
while ((line = reader.readLine()) != null) {
// 2.业务处理(词频统计)
/**
* 功能:
* 此处要进行单词的词频统计功能
* 输入:
* 每次循环读取的是一个文件,输入的是该文件的一行内容line
* 输出:
* 把每行内容line按指定分割符分割,成为一个个独立单词,进行累加统计,多个文本累计,返回结果数组
*/
}
reader.close();
in.close();
}
// 3.缓存处理结果:把统计结果写入缓存
// TODO...
// 4.将结果输出到HDFS
// 先在HDFS上创建一个空文本
FSDataOutputStream out = fs.create(new Path("output/result.txt"));
// 然后取出缓存中的内容,追加到该HDFS文本即可
// TODO...
}
}
词频统计实现
分为两步:
1)
实现上下文对象,用于保存每次的统计;2)
词频统计功能的封装调用
- 使用Map实现上下文对象
import java.util.HashMap;
import java.util.Map;
/**
* 自定义上下文对象,其实就是模仿缓存
*/
public class HDFSContext {
private Map<Object,Object> cacheMap = new HashMap<>();
// 用于从外部可以直接获取缓存
public Map<Object,Object> getCacheMap(){
return cacheMap;
}
/**
* 写数据到缓存
* @param key
* @param value
*/
public void write(Object key,Object value){
cacheMap.put(key, value);
}
/**
* 从缓存中读取数据
* @param key
* @return
*/
public Object get(Object key){
return cacheMap.get(key);
}
}
- 词频统计逻辑处理
// 自定义一个Mapper接口,封装词频统计功能
public interface HDFSMapper {
/**
* @param line 读取到的每一行数据
* @param context 上下文对象/缓存
*/
public void map(String line,HDFSContext context);
}
// 接口的功能实现
public class WordCountMapper implements HDFSMapper{
@Override
public void map(String line, HDFSContext context) {
String[] words = line.split(" "); // 按空格切割,words是一行内容的单词数组
for (String word : words) { // 遍历数组,取出每一个单词
Object value = context.get(word); // 取出缓存中的单词,
if (value == null){ // 如果value为null,则说明缓存中没有该单词
//不存在这个单词
context.write(word,1); // 第一次出现的单词,次数为1,并写入缓存
}else {
// 出现次数+1
int v = Integer.parseInt(value.toString()); // 取出单词的已经出现次数,转成int
context.write(word,v+1); // 次数+1,并写入缓存
}
}
}
}
- 调用
// 先声明类对象
HDFSContext context = new HDFSContext();
HDFSMapper mapper = new WordCountMapper();
// while里调用
while ((line = reader.readLine()) != null) {
mapper.map(line,context);
}
缓存处理结果
Map<Object,Object> contextMap = context.getCacheMap();
追加结果到HDFS
// 把Map集合转换为Set集合,进行迭代操作
Set<Map.Entry<Object, Object>> entries = contextMap.entrySet();
for (Map.Entry<Object, Object> entry : entries) {
// 取出key-value,即(word,次数),写入HDFS
out.write((entry.getKey().toString()+"\t"+entry.getValue()+"\n").getBytes());
}
System.out.println("词频统计运行成功!");
out.close();
fs.close();
完整代码
package com.hadoop.hdfs.wordcount;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Map;
import java.util.Set;
public class HDFSWordCountDemo{
public static void main(String[] args) throws Exception{
// 1.读取HDFS文件
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "hadoop");
HDFSContext context = new HDFSContext();
HDFSMapper mapper = new WordCountMapper();
// 使用Java API取出HDFS指定目录下所有要进行词频统计的单词文件,false表示不需要递归
RemoteIterator<LocatedFileStatus> files = fs.listFiles(new Path("input"), false);
// 用于循环取出多个单词文本
while (files.hasNext()) {
LocatedFileStatus file = files.next();
FSDataInputStream in = fs.open(file.getPath());
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line = null; // readLine每次读取一行
// 用于循环取出每个文本的每行内容
while ((line = reader.readLine()) != null) {
mapper.map(line,context);
}
reader.close();
in.close();
}
// 3.缓存处理结果:把统计结果写入缓存
Map<Object,Object> contextMap = context.getCacheMap();
// 4.将结果输出到HDFS
// 先在HDFS上创建一个空文本
FSDataOutputStream out = fs.create(new Path("output/result.txt"));
// 然后取出缓存中的内容,追加到该HDFS文本即可
Set<Map.Entry<Object, Object>> entries = contextMap.entrySet();
for (Map.Entry<Object, Object> entry : entries) {
out.write((entry.getKey().toString()+"\t"+entry.getValue()+"\n").getBytes());
}
System.out.println("词频统计运行成功!");
out.close();
fs.close();
}
}
查看运行结果
$ hadoop fs -cat output/*