Hadoop安裝與HDFS、MapReduce實(shí)驗(yàn)：HDFS存取代碼分析與Word Count程序代碼分析

Hadoop大數(shù)據(jù)分析——HDFS存取代碼分析

啟動(dòng)HDFS和YARN

rm -rf /opt/linuxsir/hadoop/logs/*.*
ssh root@192.168.31.132 rm -rf /opt/linuxsir/hadoop/logs/*.*
ssh root@192.168.31.133 rm -rf /opt/linuxsir/hadoop/logs/*.*

clear
cd /opt/linuxsir/hadoop/sbin
./start-dfs.sh
./start-yarn.sh
 
clear
jps
ssh root@192.168.31.132 jps
ssh root@192.168.31.133 jps

在eclipse編寫(xiě)和運(yùn)行代碼

在eclipse里面操作如下：

1. 先自定義一個(gè)項(xiàng)目 New-Java Project，名稱(chēng)自定義即可，如 java-prj
2. 接著在項(xiàng)目里面新建一個(gè)包New-Package，名稱(chēng)自定義為com.pai.hdfs_demo
3. 在包里新建一個(gè)類(lèi) New-Class，名稱(chēng)自定義為ReadWriteHDFSExample

package com.pai.hdfs_demo;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.*;
import java.nio.charset.StandardCharsets;

public class ReadWriteHDFSExample {
 // main 新建一個(gè)類(lèi)ReadWriteHDFSExample，編寫(xiě)main函數(shù)如下。main函數(shù)調(diào)用其它函數(shù)，創(chuàng)建目錄，寫(xiě)入數(shù)據(jù)，添加數(shù)據(jù)，然后再讀取數(shù)據(jù)
 public static void main(String[] args) throws IOException {
  // ReadWriteHDFSExample.checkExists();
  ReadWriteHDFSExample.createDirectory();
  ReadWriteHDFSExample.writeFileToHDFS();
  ReadWriteHDFSExample.appendToHDFSFile();
  ReadWriteHDFSExample.readFileFromHDFS();
 }

 // readFileFromHDFS 該函數(shù)讀取文件內(nèi)容，以字符串形式顯示出來(lái)
 public static void readFileFromHDFS() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
  FileSystem fileSystem = FileSystem.get(configuration);

  // Create a path
  String fileName = "read_write_hdfs_example.txt";
  Path hdfsReadPath = new Path("/javareadwriteexample/" + fileName);
  // initialize input stream
  FSDataInputStream inputStream = fileSystem.open(hdfsReadPath);
  // Classical input stream usage
  String out = IOUtils.toString(inputStream, "UTF-8");
  System.out.println(out);
  // BufferedReader bufferedReader = new BufferedReader(
  // new InputStreamReader(inputStream, StandardCharsets.UTF_8));
  // String line = null;
  // while ((line=bufferedReader.readLine())!=null){
  // System.out.println(line);
  // }
  inputStream.close();
  fileSystem.close();
 }

 // writeFileToHDFS writeFileToHDFS函數(shù)打開(kāi)文件，寫(xiě)入一行文本

public static void writeFileToHDFS() throws IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
    FileSystem fileSystem = FileSystem.get(configuration);
    // Create a path
    String fileName = "read_write_hdfs_example.txt";
    Path hdfsWritePath = new Path("/javareadwriteexample/" + fileName);
    FSDataOutputStream fsDataOutputStream = fileSystem.create(hdfsWritePath, true);
    BufferedWriter bufferedWriter = new BufferedWriter(
            new OutputStreamWriter(fsDataOutputStream, StandardCharsets.UTF_8));
    bufferedWriter.write("Java API to write data in HDFS");
    bufferedWriter.newLine();
    bufferedWriter.close();
    fileSystem.close();
}

 // appendToHDFSFile 函數(shù)打開(kāi)文件，添加一行文本。需要注意的是，需要對(duì)Configuration類(lèi)的對(duì)象configuration進(jìn)行適當(dāng)設(shè)置，否則出錯(cuò)

public static void appendToHDFSFile() throws IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
    //configuration.setBoolean("dfs.client.block.write.replace-datanode-on-failure.enabled", true);
    configuration.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
    configuration.set("dfs.client.block.write.replace-datanode-on-failure.enable","true"); 
    FileSystem fileSystem = FileSystem.get(configuration);
    // Create a path
    String fileName = "read_write_hdfs_example.txt";
    Path hdfsWritePath = new Path("/javareadwriteexample/" + fileName);
    FSDataOutputStream fsDataOutputStream = fileSystem.append(hdfsWritePath);
    BufferedWriter bufferedWriter = new BufferedWriter(
        new OutputStreamWriter(fsDataOutputStream, StandardCharsets.UTF_8));
    bufferedWriter.write("Java API to append data in HDFS file");
    bufferedWriter.newLine();
    bufferedWriter.close();
    fileSystem.close();
}

 // createDirectory 函數(shù)創(chuàng)建一個(gè)目錄
public static void createDirectory() throws IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
    FileSystem fileSystem = FileSystem.get(configuration);
    String directoryName = "/javareadwriteexample";
    Path path = new Path(directoryName);
    fileSystem.mkdirs(path);
}

 // checkExists checkExists檢查目錄或者文件是否存在。注意如下代碼的最后一個(gè)括號(hào)是ReadWriteHDFSExample類(lèi)的結(jié)束括號(hào)
public static void checkExists() throws IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
    FileSystem fileSystem = FileSystem.get(configuration);
    String directoryName = "/javareadwriteexample";
    Path path = new Path(directoryName);
    if (fileSystem.exists(path)) {
        System.out.println("File/Folder Exists : " + path.getName());
    } else {
        System.out.println("File/Folder does not Exists : " + path.getName());
    }
}

}

為了編譯通過(guò)上述Java代碼，需要把如下目錄下的jar包導(dǎo)入Eclipse項(xiàng)目的Build Path 操作序列為 右鍵點(diǎn)擊Eclipse里的Java項(xiàng)目→Properties→Java Build Path →Libraries→Add External Jars

# 添加如下路徑的包
D:hadoop-2.7.3sharehadoopcommonlib
D:hadoop-2.7.3sharehadoopcommon

D:hadoop-2.7.3sharehadoophdfs
D:hadoop-2.7.3sharehadoophdfslib


D:hadoop-2.7.3sharehadoopmapreducelib
D:hadoop-2.7.3sharehadoopmapreduce

D:hadoop-2.7.3sharehadoopyarnlib
D:hadoop-2.7.3sharehadoopyarn

在hd-master主機(jī)上檢查已經(jīng)寫(xiě)入的文件

就可以愉快地執(zhí)行了，執(zhí)行完畢上述代碼后，在hd-master主機(jī)上可以通過(guò)如下命令，檢查已經(jīng)寫(xiě)入的文件

[root@hd-master bin]# cd /opt/linuxsir/hadoop/bin
[root@hd-master bin]# ./hdfs dfs -ls /javareadwriteexample/read_write_hdfs_example.txt
-rw-r--r--   3 root supergroup         70 2024-10-10 04:47 /javareadwriteexample/read_write_hdfs_example.txt

[root@hd-master bin]# ./hdfs dfs -cat /javareadwriteexample/read_write_hdfs_example.txt
Java API to write data in HDFS
Java API to append data in HDFS file

為了多次進(jìn)行實(shí)驗(yàn)(或者為了調(diào)試代碼)，可以把HDFS文件刪除，然后再執(zhí)行或者調(diào)試Java代碼，否則一經(jīng)存在該目錄，執(zhí)行創(chuàng)建目錄的代碼就會(huì)出錯(cuò)

cd /opt/linuxsir/hadoop/bin
./hdfs dfs -rm /javareadwriteexample/*
./hdfs dfs -rmdir /javareadwriteexample

運(yùn)行完后停止YARN和HDFS

cd /opt/linuxsir/hadoop/sbin
./stop-yarn.sh
./stop-dfs.sh
 
jps
ssh root@192.168.31.132 jps
ssh root@192.168.31.133 jps

Hadoop大數(shù)據(jù)分析——Word Count程序代碼分析

在eclipse編寫(xiě)和運(yùn)行代碼

package mywordcount;

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {
    //定義WordCount類(lèi)的內(nèi)部類(lèi)TokenizerMapper 該類(lèi)實(shí)現(xiàn)了map函數(shù)，把從文件讀取的每個(gè)word變成一個(gè)形式為<word,1>的Key Value對(duì)，輸出到map函數(shù)的參數(shù)context對(duì)象，由執(zhí)行引擎完成Shuffle
 public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

  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 {
   StringTokenizer itr = new StringTokenizer(value.toString());
   while (itr.hasMoreTokens()) {
    word.set(itr.nextToken());
    context.write(word, one);
   }
  }
 }
    //定義WordCount類(lèi)的內(nèi)部類(lèi)IntSumReducer    該類(lèi)實(shí)現(xiàn)了reduce函數(shù)，它收攏所有相同key的、形式為<word,1>的Key-Value對(duì)，對(duì)Value部分進(jìn)行累加，輸出一個(gè)計(jì)數(shù)
 public static class IntSumReducer extends
 Reducer<Text, IntWritable, Text, IntWritable> {
  private IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, Context context)
    throws IOException, InterruptedException {
   int sum = 0;
   for (IntWritable val : values) {
    sum += val.get();
   }
   result.set(sum);
   context.write(key, result);
   String thekey = key.toString();
   int thevalue = sum;
  }
 }
    // WordCount類(lèi)的main函數(shù)，負(fù)責(zé)配置Job的若干關(guān)鍵的參數(shù)，并且啟動(dòng)這個(gè)Job。在main函數(shù)中，conf對(duì)象包含了一個(gè)屬性即“fs.defaultFS”，它的值為“hdfs://192.168.31.131:9000”，使得WordCount程序知道如何存取HDFS

 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
  if (otherArgs.length != 2) {
   System.err.println("Usage: wordcount <in> <out>");
   System.exit(2);
  }
  conf.set("fs.defaultFS", "hdfs://192.168.31.131:9000");
  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.exit(job.waitForCompletion(true) ? 0 : 1);
 }

}

在hd-master主機(jī)上檢查已經(jīng)寫(xiě)入的文件

[root@hd-master bin]# ./hdfs dfs -ls /output1
Found 2 items
-rw-r--r--   3 root supergroup          0 2024-10-10 05:17 /output1/_SUCCESS
-rw-r--r--   3 root supergroup         89 2024-10-10 05:17 /output1/part-r-00000

 [root@hd-master bin]# ./hdfs dfs -cat /output1/part-r-00000
I       1
apache  1
cloudera        1
google  1
hadoop  8
hortonworks     1
ibm     1
intel   1
like    1
microsoft       1