需求:
统计hdfs上某个文件中词频个数,并将统计结果存储到hdfs上的另一个文件中
需求分析:
连接hdfs,从hsfs上读取文件,按行读取;
将读取到的数据按空格切割字符串,分割单词;
遍历分割后的字符串数组,使用map函数去重,统计词频个数;
遍历map中数据,循环将每一条数据写入hdfs上的某个文件中;
在hsfs上使用 hadoop fs -ls 目录/文件名查看是否写入成功。
代码部分:
public class WordCount{
public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
HashMap<String, Integer> map = new HashMap<>();
//连接hdfs
FileSystem.getFileSystem fs = FileSystem.get(new URI("hdfs://hadoop01:9000"), new Configuration(), "root");
//使用流数据读取hdfs上的文件
FSDataInputStream inputStream = fs.open(new Path("/ghm.txt"));
//为了实现按行读取,使用缓冲流读取文件
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
String line = null;
//循环读取每一行文件
while((line=reader.readLine())!=null) {
根据空格或tab分割字符串,分割出每个单词
String[] split = line.split("\\s");
//循环遍单词数组使用map函数去重并统计词频个数
for (String word : split) {
Integer count = map.getOrDefault(word, 0);
count++;
map.put(word, count);
}
}
//指定文件要写入的位置
FSDataOutputStream create = fs.create(new Path("/part-r-0001"));
//循环遍历map中数据,将map中数据写入hdfs
Set<Entry<String, Integer>> entrySet = map.entrySet();
for (Entry<String, Integer> entry : entrySet) {
//执行写入操作 create.write((entry.getKey()+"="+entry.getValue()+"\r\n").getBytes());
System.out.println(entry);
}
//关闭资源
create.close();
reader.close();
inputStream.close();
fs.close();
}
}