Hadoop多個檔案單詞計數
阿新 • • 發佈：2019-01-07
Hadoop的安裝
首先下載Hadoop的安裝包，這裡使用2.7.3版本。解壓到/usr/local下
sudo tar -zxvf hadoop-2.7.3.tar.gz -C /usr/local/ 然後更改hadoop-2.7.3的屬主
sudo chown -R jessin /usr/local/hadoop-2.7.3在/etc/profile下新增HADOOP_HOME環境變數，並將其bin放到PATH路徑下：
export HADOOP_HOME=/usr/local/hadoop-2.7.3
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME} 接著使配置立即生效：
source /etc/profile 接下來在$HADOOP_HOME/etc/hadoop下修改或者新增四個檔案，這時使用的是偽分散式，還是會使用HDFS。
core-site.xml
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. --> hdfs-site.xml
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>dfs.replication</name>
<value>1</value>
</property>
<property>
<name>dfs.namenode.name.dir</name>
<value>file:/usr/local/hadoop-2.7.3/tmp/dfs/name</value>
</property>
<property>
<name>dfs.datanode.data.dir</name>
<value>file:/usr/local/hadoop-2.7.3/tmp/dfs/data</value>
</property>
</configuration>mapred-site.xml
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>mapreduce.framework.name</name>
<value>yarn</value>
</property>
</configuration>yarn-site.xml
<?xml version="1.0"?>
<configuration>
<!-- Site specific YARN configuration properties -->
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
</configuration>必須在$HADOOP_HOME/etc/hadoop/hadoop-env.xml下新增JAVA_HOME的環境變數：
export JAVA_HOME=/usr/lib/jvm/jdk1.7.0Hadoop的啟動和終止
格式化HDFS
hdfs namenode -format啟動hdfs和yarn
start-dfs.sh
start-yarn.sh在hdfs下建立一個HOME資料夾
hadoop fs -mkdir /user/jessin結束Hadoop
stop-yarn.sh
stop-dfs.sh單詞計數程式
單詞計數程式是Hadoop的hello world程式，這裡使用Maven來構建，需要在pom.xml新增如下兩個jar:
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>2.7.3</version>
</dependency>WordCount.java。需要注意的是Mapper和Reducer的泛型引數的前兩個是輸入的key,value型別，後兩個是輸出的key,value型別。
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 {
//TokenizerMapper繼承Mapper類，並重寫其map方法。
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 {
//map方法中的value值儲存的是文字檔案中的一行資訊（以回車符為行結束標記）。
//map方法中的key值為該行的首字元相對與文字檔案的首地址的偏移量。
//StringTokenizer類將每一行拆分成一個個的單詞，並將<word,1>作為map方法的結果輸出。
//其中IntWritable和Text類是Hadoop對int和String類的序列化封裝，這些類能夠被序列化，以便在分散式環境中進行資料交換。
StringTokenizer itr = new StringTokenizer(value.toString());
System.out.println("key : " + key + " value : " + value);
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);//輸出<key,value>為<word,one>
}
}
}
//IntSumReducer繼承Reducer類，並重寫其reduce方法。
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 {
//reduce方法的輸入引數key為單個單詞;
//而Iterable<IntWritable> values為各個Mapper上對應單詞的計數值所組成的列表。
int sum = 0;
for (IntWritable val : values) {//遍歷求和
sum += val.get();
}
result.set(sum);
context.write(key, result);//輸出求和後的<key,value>
}
}
//在MapReduce中，由Job物件負責管理和執行一個計算任務，並通過Job的一些方法對任務的引數進行相關的設定。
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);
}
Job job = new Job(conf, "word count");
job.setJarByClass(WordCount.class);
//使用TokenizerMapper類完成Map過程；
job.setMapperClass(TokenizerMapper.class);
//使用IntSumReducer類完成Combiner過程；
job.setCombinerClass(IntSumReducer.class);
//使用IntSumReducer類完成Reducer過程；
job.setReducerClass(IntSumReducer.class);
//設定了Map過程和Reduce過程的輸出型別，其中設定key的輸出型別為Text；
job.setOutputKeyClass(Text.class);
//設定了Map過程和Reduce過程的輸出型別，其中設定value的輸出型別為IntWritable；
job.setOutputValueClass(IntWritable.class);
//設定任務資料的輸入路徑；
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
//設定任務輸出資料的儲存路徑；
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
//呼叫job.waitForCompletion(true) 執行任務，執行成功後退出；
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}可以在根目錄下打jar
mvn clean compile package 結果如下：
然後啟動hadoop，在hdfs建立wordcount/input資料夾，將程式碼目錄的的兩個輸入檔案上傳到hdfs
hadoop fs -mkdir -p wordcount/input
hadoop fs -put src/main/resources/input wordcount
hadoop fs -ls wordcount 結果如下：
在執行job之前，在HDFS上的輸出資料夾必須不存在，否則會執行失敗。可以使用以下命令刪除HDFS上的資料夾。預設是使用者的HOME目錄：
hadoop fs -rm -r wordcount/output使用以下命令將job提交到hadoop執行，注意最後兩個是HDFS上的輸入和輸出檔案，倒數第三個WordCount是jar中的主函式：
hadoop jar ~/Documents/Program/final/hadoop_helloworld/target/hadoop-helloworld-1.0-SNAPSHOT.jar WordCount wordcount/input wordcount/output 結果如下：
在hdfs的wordcount/output下生成了兩個檔案，其中part-r-00000含有輸出結果
hadoop fs -ls wordcount/output
hadoop fs -cat wordcount/output/part-r-00000 執行結果如下：
