1. 云图网首页
  2. 技术专区
  3. 大数据

Hadoop综合练习第四节–MapReduce计算气象温度等例子详解大数据

• 大数据

1    运行环境说明… 4

1.1     硬软件环境… 4

1.2     机器网络环境… 4

2    书面作业1：对云计算的看法… 4

2.1     书面作业1内容… 4

2.2     回答… 5

3    书面作业2：使用MapReduce求每年最低温度… 6

3.1     书面作业2内容… 6

3.2     运行代码… 6

3.2.1   MinTemperature. 6

3.2.2   MinTemperatureMapper6

3.2.3   MinTemperatureReducer7

3.3     实现过程… 8

3.3.1   编写代码… 8

3.3.2   编译代码… 9

3.3.3   打包编译文件… 9

3.3.4   创建目录… 10

3.3.5   解压气象数据并上传到HDFS中… 10

3.3.6   运行程序… 12

3.3.7   查看结果… 12

3.3.8   通过页面结果… 12

4    书面作业3：求温度平均值能使用combiner吗？… 15

4.1     书面作业3内容… 15

4.2     回答… 15

4.3     程序代码… 15

4.3.1   AvgTemperature.java. 15

4.3.2   AvgTemperatureMapper.java. 16

4.3.3   AvgTemperatureCombiner.java. 17

4.3.4   AvgTemperatureReducer.java. 17

4.4     实现过程… 18

4.4.1   编写代码… 18

4.4.2   编译代码… 20

4.4.3   打包编译文件… 20

4.4.4   运行程序… 21

4.4.5   查看结果… 22

5    书面作业4：使用Hadoop流求最高温度（awk脚本）… 22

5.1     书面作业4内容… 22

5.2     程序代码… 22

5.2.1   执行代码… 22

5.2.2   mapper.sh. 22

5.2.3   reducer.sh. 23

5.3     实现过程… 23

5.3.1   编写代码… 23

5.3.2   运行程序… 24

5.3.3   查看结果… 24

6    书面作业4：使用Hadoop流求最高温度（Python语言）… 25

6.1     书面作业4内容… 25

6.2     程序代码… 25

6.2.1   执行代码… 25

6.2.2   mapper.py. 25

6.2.3   reducer.py. 25

6.3     实现过程… 26

6.3.1   编写代码… 26

6.3.2   运行程序… 27

6.3.3   查看结果… 27

7    书面作业5：MapReduce是否可以自动识别新增节点？… 28

7.1     书面作业5内容… 28

7.2     程序代码… 28

7.3     实现过程… 28

7.3.1   环境准备… 28

7.3.2   准备数据… 32

7.3.3   不启动Hadoop4，运行任务… 33

7.3.4   启动Hadoop4，运行任务… 34

7.3.5   重启Hadoop集群，运行任务… 35

7.3.6   结论… 37

8    书面作业6：使用Hadoop公平调度器… 38

8.1     书面作业6内容… 38

8.2     程序代码… 38

8.3     实现过程… 38

8.3.1   打开公平调度器… 38

8.3.2   验证启动公平调度器… 39

8.3.3   准备数据… 40

8.3.4   运行分析MapReduce. 41

8.3.5   观察结果… 42

9    问题解决… 43

9.1     在作业5中新增节点后，DataNode无法启动… 43

 

1 运行环境说明

1.1  硬软件环境

l  主机操作系统：Windows 64 bit，双核4线程，主频2.2G，6G内存

l  虚拟软件：VMware® Workstation 9.0.0 build-812388

l  虚拟机操作系统：CentOS 64位，单核，1G内存

l  JDK：1.7.0_55 64 bit

l  Hadoop：1.1.2

1.2  机器网络环境

集群包含三个节点：1个namenode、2个datanode，其中节点之间可以相互ping通。节点IP地址和主机名分布如下：

序号

IP地址

机器名

类型

用户名

运行进程

1

10.88.147.221

hadoop1

名称节点

hadoop

NN、SNN、JobTracer

2

10.88.147.222

hadoop2

数据节点

hadoop

DN、TaskTracer

3

10.88.147.223

hadoop3

数据节点

hadoop

DN、TaskTracer

所有节点均是CentOS6.5 64bit系统，防火墙均禁用，所有节点上均创建了一个hadoop用户，用户主目录是/usr/hadoop。所有节点上均创建了一个目录/usr/local/hadoop，并且拥有者是hadoop用户。

2 书面作业1：对云计算的看法

2.1  书面作业1内容

说说你对云计算的看法，是忽悠？还是能带来真实价值的东西？

2.2  回答

云计算是对现有资源集中优化后，对客户提供服务，从现在的情况来看云计算真实的为大家提供了服务，比如：网盘等。至于云计算更为准确的定义为美国国家标准与技术研究院（NIST）定义：云计算是一种按使用量付费的模式，这种模式提供可用的、便捷的、按需的网络访问，进入可配置的计算资源共享池（资源包括网络，服务器，存储，应用软件，服务），这些资源能够被快速提供，只需投入很少的管理工作或与服务供应商进行很少的交互。

云计算特点如下：

(1) 超大规模：“云”具有相当的规模，赋予用户前所未有的计算能力；

(2) 虚拟化：云计算支持用户在任意位置、使用各种终端获取应用服务；

(3) 高可靠性：“云”使用了数据多副本容错、计算节点同构可互换等措施来保障服务的高可靠性；

(4) 通用性：云计算不针对特定的应用，同一个“云”可以同时支撑不同的应用运行；

(5) 高可扩展性：“云”的规模可以动态伸缩，满足应用和用户规模增长的需要；

(6) 按需服务：“云”是一个庞大的资源池，可以需购买；

(7) 极其廉价：由于“云”的特殊容错措施可以采用极其廉价的节点来构成云，“云”的自动化集中式管理使大量企业无需负担日益高昂的数据中心管理成本；

(8) 潜在的危险性

云计算可以认为包括以下几个层次的服务：基础设施即服务（IaaS），平台即服务（PaaS）和软件即服务（SaaS）。

l  IaaS(Infrastructure-as-a-Service)：基础设施即服务。消费者通过Internet可以从完善的计算机基础设施获得服务。例如：硬件服务器租用。

l  PaaS：PaaS(Platform-as-a- Service)：平台即服务。PaaS实际上是指将软件研发的平台作为一种服务，以SaaS的模式提交给用户。因此，PaaS也是SaaS模式的一种应用。但是，PaaS的出现可以加快SaaS的发展，尤其是加快SaaS应用的开发速度。例如：软件的个性化定制开发。

l  SaaS：SaaS(Software-as-a- Service)：软件即服务。它是一种通过Internet提供软件的模式，用户无需购买软件，而是向提供商租用基于Web的软件，来管理企业经营活动。例如：阳光云服务器。

3  书面作业2：使用MapReduce求每年最低温度

3.1  书面作业2内容

下载气象数据集部分数据，写一个Map-Reduce作业，求每年的最低温度，部署并运行之，抓图过程

3.2  运行代码

3.2.1MinTemperature

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.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

publicclass MinTemperature {

   

    publicstaticvoid main(String[] args) throws Exception {

        if(args.length != 2) {

            System.err.println("Usage: MinTemperature<input path> <output path>");

            System.exit(-1);

        }

       

        Job job = new Job();

        job.setJarByClass(MinTemperature.class);

        job.setJobName("Min temperature");

        FileInputFormat.addInputPath(job, new Path(args[0]));

        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(MinTemperatureMapper.class);

        job.setReducerClass(MinTemperatureReducer.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

 

 

3.2.2MinTemperatureMapper

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

 

publicclass MinTemperatureMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

 

    privatestatic final intMISSING = 9999;

   

    @Override

    publicvoid map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

       

        String line = value.toString();

        String year = line.substring(15, 19);

       

        int airTemperature;

        if(line.charAt(87) == '+') {

            airTemperature = Integer.parseInt(line.substring(88, 92));

        } else {

            airTemperature = Integer.parseInt(line.substring(87, 92));

        }

       

        String quality = line.substring(92, 93);

        if(airTemperature != MISSING && quality.matches("[01459]")) {

            context.write(new Text(year), new IntWritable(airTemperature));

        }

    }

}

3.2.3MinTemperatureReducer

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

publicclass MinTemperatureReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

 

    @Override

    publicvoid reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

       

        int minValue = Integer.MAX_VALUE;

        for(IntWritable value : values) {

            minValue = Math.min(minValue, value.get());

        }

        context.write(key, new IntWritable(minValue));

    }

}

3.3  实现过程

3.3.1编写代码

进入/usr/local/hadoop-1.1.2/myclass目录，在该目录中建立MinTemperature.java、MinTemperatureMapper.java和MinTemperatureReducer.java代码文件，代码内容为3.2所示，执行命令如下：

cd /usr/local/hadoop-1.1.2/myclass/

vi MinTemperature.java

vi MinTemperatureMapper.java

vi MinTemperatureReducer.java

clip_image002

MinTemperature.java：

clip_image004

MinTemperatureMapper.java：

clip_image006

MinTemperatureReducer.java:

clip_image008

3.3.2编译代码

在/usr/local/hadoop-1.1.2/myclass目录中，使用如下命令对java代码进行编译，为保证编译成功，加入classpath变量，引入hadoop-core-1.1.2.jar包：

javac -classpath ../hadoop-core-1.1.2.jar *.java

ls

clip_image010

3.3.3打包编译文件

把编译好class文件打包，否则在执行过程会发生错误。把打好的包移动到上级目录并删除编译好的class文件：

jar cvf ./MinTemperature.jar ./*.class

ls

mv *.jar ..

rm *.class

clip_image012

3.3.4创建目录

进入/usr/local/hadoop-1.1.2/bin目录，在HDFS中创建气象数据存放路径/usr/hadoop/in，执行命令如下：

cd /usr/local/hadoop-1.1.2/bin

hadoop fs -mkdir /usr/hadoop/in

hadoop fs -ls /usr/hadoop                

clip_image014

3.3.5解压气象数据并上传到HDFS中

使用SSH工具（参见第1、2周2.1.3.1Linux文件传输工具所描述）把从NCDC下载的气象数据上传到上步骤创建的目录/usr/local/hadoop-1.1.2/input中。

clip_image016

使用zcat命令把这些数据文件解压并合并到一个sample.txt文件中，合并后把这个文件上传到HDFS文件系统的/usr/hadoop/in目录中：

cd /usr/local/hadoop-1.1.2/input

zcat *.gz > sample.txt

hadoop fs -copyFromLocal sample.txt /usr/hadoop/in

clip_image018

clip_image020

气象数据具体的下载地址为 ftp://ftp3.ncdc.noaa.gov/pub/data/noaa/ ，该数据包括1900年到现在所有年份的气象数据，大小大概有70多个G。为了测试简单，我们这里选取一部分的数据进行测试

3.3.6运行程序

以jar的方式启动MapReduce任务，执行输出目录为/usr/hadoop/out：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/sample.txt  /usr/hadoop/out

clip_image022

3.3.7查看结果

执行成功后，查看/usr/hadoop/out目录中是否存在运行结果，使用cat查看结果：

hadoop fs -ls /usr/hadoop/out

hadoop fs -cat /usr/hadoop/out/part-r-00000

clip_image024

3.3.8通过页面结果

1.     查看jobtracker.jsp

http://10.88.147.221:50030/jobtracker.jsp

clip_image026

已经完成的作业任务：

clip_image028

任务的详细信息：

clip_image030

clip_image032

2.     查看dfshealth.jsp

http://10.88.147.221:50070/dfshealth.jsp

clip_image034

分别查看HDFS文件系统和日志

clip_image036

clip_image038

 

4    书面作业3：求温度平均值能使用combiner吗？

4.1  书面作业3内容

（选作）如果求温度的平均值，能使用combiner吗？有没有变通的方法？说说你的看法

4.2  回答

不能使用，因为求平均值和前面求最值存在差异，各局部最值的最值还是等于整体的最值的，但是对于平均值而言，各局部平均值的平均值将不再是整体的平均值了，所以不能用combiner。可以通过变通的办法使用combiner来计算平均值，即在combiner的键值对中不直接存储最后的平均值，而是存储所有值的和个数，最后在reducer输出时再用和除以个数得到平均值。

4.3  程序代码

4.3.1AvgTemperature.java

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.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

publicclass AvgTemperature {

   

    publicstaticvoid main(String[] args) throws Exception {

         

        if(args.length != 2) {

            System.out.println("Usage: AvgTemperatrue <input path><output path>");

            System.exit(-1);

        }

       

        Job job = new Job();

        job.setJarByClass(AvgTemperature.class);

        job.setJobName("Avg Temperature");

        FileInputFormat.addInputPath(job, new Path(args[0]));

        FileOutputFormat.setOutputPath(job, new Path(args[1]));

       

        job.setMapperClass(AvgTemperatureMapper.class);

        job.setCombinerClass(AvgTemperatureCombiner.class);

        job.setReducerClass(AvgTemperatureReducer.class);

       

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(Text.class);

       

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

       

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

4.3.2AvgTemperatureMapper.java

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

 

publicclass AvgTemperatureMapper extends Mapper<LongWritable, Text, Text, Text> {

 

    privatestaticfinalintMISSING = 9999;

   

    @Override

    publicvoid map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{

       

        String line = value.toString();

        String year = line.substring(15, 19);

       

        int airTemperature;

        if(line.charAt(87) == '+') {

            airTemperature = Integer.parseInt(line.substring(88, 92));

        } else {

            airTemperature =  Integer.parseInt(line.substring(87, 92));

        }

       

        String quality = line.substring(92, 93);

        if(airTemperature != MISSING && !quality.matches("[01459]")) {

            context.write(new Text(year), new Text(String.valueOf(airTemperature)));

        }

    }

}

4.3.3AvgTemperatureCombiner.java

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

publicclass AvgTemperatureCombiner extends Reducer<Text, Text, Text, Text>{

 

    @Override

    publicvoid reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

       

        double sumValue = 0;

        long numValue = 0;

       

        for(Text value : values) {

            sumValue += Double.parseDouble(value.toString());

            numValue ++;

        }

       

        context.write(key, new Text(String.valueOf(sumValue) + ',' + String.valueOf(numValue)));

    }

}

4.3.4AvgTemperatureReducer.java

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

publicclass AvgTemperatureReducer extends Reducer<Text, Text, Text, IntWritable>{

 

    @Override

    publicvoid reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

       

        double sumValue = 0;

        long numValue = 0;

        int avgValue = 0;

       

        for(Text value : values) {

            String[] valueAll = value.toString().split(",");

            sumValue += Double.parseDouble(valueAll[0]);

            numValue += Integer.parseInt(valueAll[1]);

        }

       

        avgValue  = (int)(sumValue/numValue);

        context.write(key, new IntWritable(avgValue));

    }

}

4.4  实现过程

4.4.1编写代码

进入/usr/local/hadoop-1.1.2/myclass目录，在该目录中建立AvgTemperature.java、AvgTemperatureMapper.java、AvgTemperatureCombiner.java和AvgTemperatureReducer.java代码文件，代码内容为4.3所示，执行命令如下：

cd /usr/local/hadoop-1.1.2/myclass/

vi AvgTemperature.java

vi AvgTemperatureMapper.java

vi AvgTemperatureCombiner.java

vi AvgTemperatureReducer.java

clip_image040

AvgTemperature.java：

clip_image042

AvgTemperatureMapper.java：

clip_image044

AvgTemperatureCombiner.java：

clip_image046

AvgTemperatureReducer.java:

clip_image048

4.4.2编译代码

在/usr/local/hadoop-1.1.2/myclass目录中，使用如下命令对java代码进行编译，为保证编译成功，加入classpath变量，引入hadoop-core-1.1.2.jar包：

javac -classpath ../hadoop-core-1.1.2.jar *.java

ls

clip_image050

4.4.3打包编译文件

把编译好class文件打包，否则在执行过程会发生错误。把打好的包移动到上级目录并删除编译好的class文件：

jar cvf ./AvgTemperature.jar ./*.class

ls

mv *.jar ..

rm *.class

clip_image052

4.4.4运行程序

数据使用作业2求每年最低温度的气象数据，数据在HDFS位置为/usr/ hadoop/in/sample.txt，以jar的方式启动MapReduce任务，执行输出目录为/usr/hadoop/out1：

cd /usr/local/hadoop-1.1.2

hadoop jar AvgTemperature.jar AvgTemperature /usr/hadoop/in/sample.txt  /usr/hadoop/out1

clip_image054

clip_image056

4.4.5查看结果

执行成功后，查看/usr/hadoop/out目录中是否存在运行结果，使用cat查看结果：

hadoop fs -ls /usr/hadoop/out1

hadoop fs -cat /usr/hadoop/out1/part-r-00000

clip_image058

5    书面作业4：使用Hadoop流求最高温度（awk脚本）

5.1  书面作业4内容

（选作）使用hadoop流的方法来实现对气象数据集求最高温度的分析任务（可能要使用awk这类脚本工具）

5.2  程序代码

5.2.1执行代码

hadoop jar contrib/streaming/hadoop-streaming-1.1.2.jar /

-input /usr/hadoop/in/sample.txt /

-output /usr/hadoop/out_awk /

-mapper myclass/map.sh /

-reducer myclass/reducer.sh /

-file myclass/map.sh /

-file myclass/reducer.sh

5.2.2mapper.sh

#!/usr/bin/awk -f

BEGIN { FS = "," }

{ year = substr($0, 16, 4) + 0;

  temp = substr($0, 88, 5) + 0;

  q = substr($0, 93, 1);

  LF="/n";

  if(temp != 9999 && q ~ /[01459]/)

  {

    printf "%s %s %s",  year, temp,LF;

  }

}

END {

    printf "/n"

}

5.2.3reducer.sh

#!/usr/bin/awk -f

{

  temp[$1] =  $2

  if ( temp[$1] > maxtemp[$1] )

    maxtemp[$1] = temp[$1];

 

}

END {

  for(num in maxtemp)

    printf "%s %s/n", num, maxtemp[num];

}

5.3  实现过程

5.3.1编写代码

进入/usr/local/hadoop-1.1.2/myclass目录，在该目录中建立mapper.sh和reducer.sh代码文件，代码内容为5.2所示，执行命令如下：

cd /usr/local/hadoop-1.1.2/myclass/

vi mapper.sh

vi reducer.sh

clip_image060

mapper.sh：

clip_image062

reducer.sh：

clip_image064

5.3.2运行程序

数据使用作业2求每年最低温度的气象数据，数据在HDFS位置为/usr /hadoop/in/sample.txt，启动MapReduce任务，执行输出目录为/usr/hadoop/out_awk：

cd /usr/local/hadoop-1.1.2

hadoop jar contrib/streaming/hadoop-streaming-1.1.2.jar /

-input /usr/hadoop/in/sample.txt /

-output /usr/hadoop/out_awk /

-mapper myclass/mapper.sh /

-reducer myclass/reducer.sh /

-file myclass/mapper.sh /

-file myclass/reducer.sh

clip_image066

5.3.3查看结果

执行成功后，查看/usr/hadoop/out_awk目录中是否存在运行结果，使用cat查看结果：

hadoop fs -ls /usr/hadoop/out_awk

hadoop fs -cat /usr/hadoop/out_awk/part-00000

clip_image068

6    书面作业4：使用Hadoop流求最高温度（Python语言）

6.1  书面作业4内容

（选作）使用hadoop流的方法来实现对气象数据集求最高温度的分析任务（可能要使用awk这类脚本工具）

6.2  程序代码

6.2.1执行代码

hadoop jar contrib/streaming/hadoop-streaming-1.1.2.jar  /

-input /usr/hadoop/in/sample.txt /

-output /usr/hadoop/out_python /

-mapper myclass/mapper.py /

-reducer myclass/reduce.py /

-file myclass/mapper.py /

-file myclass/reduce.py

6.2.2mapper.py

#!/usr/bin/env python

 

import re

import sys

 

for line in sys.stdin:

    val = line.strip()

    (year, temp, q) = (val[15:19], val[87:92], val[92:93])

    if (temp != "+9999" and re.match("[01459]", q)):

        print "%s/t%s" % (year, temp)

6.2.3reducer.py

#!/usr/bin/env python

 

import sys

 

(last_key, max_val) = (None, -sys.maxint)

for line in sys.stdin:

    (key, val) = line.strip().split("/t")

    if last_key and last_key != key:

        print "%s/t%s" % (last_key, max_val)

        (last_key, max_val) = (key, int(val))

    else:

        (last_key, max_val) = (key, max(max_val, int(val)))

 

if last_key:

    print "%s/t%s" % (last_key, max_val)

6.3  实现过程

6.3.1编写代码

进入/usr/local/hadoop-1.1.2/myclass目录，在该目录中建立mapper.py和reducer.py代码文件，代码内容为6.2所示，执行命令如下：

cd /usr/local/hadoop-1.1.2/myclass/

vi mapper.py

vi reducer.py

clip_image070

mapper.py：

clip_image072

reducer.py：

clip_image074

6.3.2运行程序

数据使用作业2求每年最低温度的气象数据，数据在HDFS位置为/usr/hadoop/in/sample.txt，启动MapReduce任务，执行输出目录为/usr/hadoop/out_py：

cd /usr/local/hadoop-1.1.2

hadoop jar contrib/streaming/hadoop-streaming-1.1.2.jar /

-input /usr/hadoop/in/sample.txt /

-output /usr/hadoop/out_py /

-mapper myclass/mapper.py /

-reducer myclass/reducer.py /

-file myclass/mapper.py /

-file myclass/reducer.py

clip_image076

6.3.3查看结果

执行成功后，查看/usr/hadoop/out_py目录中是否存在运行结果，使用cat查看结果：

hadoop fs -ls /usr/hadoop/out_py

hadoop fs -cat /usr/hadoop/out_py/part-00000

clip_image078

 

7    书面作业5：MapReduce是否可以自动识别新增节点？

7.1  书面作业5内容

（选作）如果向正在运行的Hadoop集群增加一个新节点，Map-Reduce体系是否可以自动识别并使用这个新节点？如果不能怎样才可以将其加入到Map-Reduce体系？用实验进行验证

7.2  程序代码

参见作业2代码并打成jar包

7.3  实现过程

在这里将运行本周作业2求每年最低温度的MapReduce（当然也可以运行求最高温度等例子），为了能够更清楚的观察实验的效果，在该作业中运行的数据尽可能大，以下实验中将使用1970~1972年各年份前400个气象站数据，解压后大概480M左右。在本实验中将观察两个情况：一种是运行过程新增节点，另一种是重启集群观察作业是否运行在新增节点上？

新增节点信息：

序号

IP地址

机器名

类型

用户名

运行进程

4

10.88.147.224

Hadoop4

数据节点

hadoop

DN、TaskTracer

7.3.1环境准备

参考第三周第二题第四个问题–SNN与NN分离—进行环境准备，增加一个新节点并确保各节点能够免密码SSH访问

7.3.1.1复制虚拟机

复制DataNode节点所在虚拟机，新虚拟机作为新增节点

clip_image080

7.3.1.2设置新增节点虚拟机IP地址

设置该节点IP地址为：10.88.147.224

clip_image082

7.3.1.3设置新增节点虚拟机名称

设置新增节点虚拟机名称为：hadoop4

sudo vi /etc/sysconfig/network

clip_image084

clip_image086

7.3.1.4所有节点hosts 文件加入新增节点的 IP对应信息

在所有节点/etc/hosts文件中加入新增节点的IP地址10.88.147.224对应hadoop4

sudo vi /etc/hosts

clip_image088

7.3.1.5所有节点slavers文件加入新增节点信息

在新增节点hadoop4机器的slaves文件中加入新增节点机器名信息，使用如下命令：

sudo vi /usr/local/hadoop-1.1.2/conf/slaves

在slaves文件中加入新增节点机器名

clip_image090

7.3.1.6重启所有虚拟机

clip_image092

7.3.1.7配置ssh免密码登录

1.     在hadoop4（10.88.147.244）节点中使用ssh-keygen -t rsa生成私钥和公钥；

clip_image094

2.     把hadoop4（10.88.147.244）节点中公钥信息加入到authorized_keys文件中；

chmod 400 -R /home/hadoop/.ssh

cat id_rsa.pub >> authorized_keys

cat authorized_keys

clip_image096

3.     把authorized_keys分发到各个节点上；

scp authorized_keys [email protected]:/home/hadoop/.ssh

clip_image098

4.     验证是否能够免登录到各个节点；

clip_image100

7.3.2准备数据

参考作业2，为了能够更清楚的观察实验的效果，在该作业中运行的数据尽可能大，在这里上传数据大概480M左右。使用SSH工具（参见第1、2周2.1.3.1    Linux文件传输工具所描述）把从NCDC下载的气象数据上传到上步骤创建的目录/usr/local/hadoop-1.1.2/input中。

clip_image102

先把这三年的数据放到/usr/local/hadoop-1.1.2/input目录下，然后使用zcat命令把这些数据文件解压并合并到一个newnode.txt文件中，合并后把这个文件上传到HDFS文件系统的/usr/hadoop/in目录中：

cd /usr/local/hadoop-1.1.2/input

mv 1971/*.gz .

mv 1972/*.gz .

mv 1973/*.gz .

zcat *.gz > newnode.txt

ll newnode.txt

clip_image104

hadoop fs -copyFromLocal newnode.txt /usr/hadoop/in

hadoop fs -ls /usr/hadoop/in

clip_image106

7.3.3不启动Hadoop4，运行任务

使用本周作业2打包的jar包，启动作业，执行输出目录为/usr/hadoop/out_newnode1：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/newnode.txt  /usr/hadoop/out_newnode1

clip_image108

查看运行作业节点信息，Hadoop集群由两个DataNode节点，分别为Hadoop2和Hadoop3：

clip_image110

7.3.4启动Hadoop4，运行任务

再次启动作业，执行输出目录为/usr/hadoop/out_newnode2：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/newnode.txt  /usr/hadoop/out_newnode2

clip_image112

运行作业后，启动新增节点Hadoop4中的DataNode和TaskTracker进程

cd /usr/local/hadoop-1.1.2/bin

hadoop-daemon.sh start datanode

hadoop-daemon.sh start tasktracker

jps

clip_image114

观察结果，可以看到Hadoop集群识别到Hadoop4节点并发现该节点运行的TaskTracker：

clip_image116

clip_image118

在新增节点Hadoop4正在运行task任务

clip_image120

 

7.3.5重启Hadoop集群，运行任务

停止Hadoop集群运行：

cd /usr/local/hadoop-1.1.2/bin

stop-all.sh

clip_image122

在Hadoop4节点中结束DataNode和TaskTracker进程：

cd /usr/local/hadoop-1.1.2/bin

hadoop-daemon.sh stop datanode

hadoop-daemon.sh stop tasktracker

clip_image124

在Hadoop1节点中HADOOP_HOME/conf/salvers中加入Hadoop4节点

cd /usr/local/hadoop-1.1.2/conf

sudo vi /usr/local/hadoop-1.1.2/conf/slaves

clip_image126

启动Hadoop集群，从下图可以看见Hadoop4节点已经随Hadoop集群启动：

cd /usr/local/hadoop-1.1.2/bin

start-all.sh

clip_image128

clip_image130

再次启动运行作业，可以观察到Hadoop4节点得到运行任务：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/newnode.txt  /usr/hadoop/out_newnode3

clip_image132

clip_image134

 

7.3.6结论

综上，当Hadoop集群有新节点加入时，正在运行的MapReduce作业自动识别并使用新节点；当新节点加入Hadoop集群后，再启动MapReduce作业，MapReduce作业也能自动识别并使用新节点。

8    书面作业6：使用Hadoop公平调度器

8.1  书面作业6内容

（选作）怎样打开Hadoop的公平调度器？怎样证实当前处于公平调度下。设计一个实验方案验证FIFO调度器和公平调度器的异同，最好能实践之

8.2  程序代码

参见作业2代码并打成jar包

8.3  实现过程

在这里将运行本周作业2求每年最低温度的MapReduce（当然也可以运行求最高温度等例子），为了能够更加明显看到实验效果该实验将准备两份实验数据：数据A为1970~1972年各年份前10个气象站合并数据，数据B为1970~1972年各年份前240个气象站合并数据，那么分析数据A的MapReduceA时间将大大少于分析数据B的MapReduceB时间。在实验过程中将让分析数据B的MapReduceB先启动，然后启动分析数据A的MapReduceA，观察MapReduceA和MapReduceB哪一个先结束？

8.3.1打开公平调度器

要让公平调度器能在Hadoop中运行，需要把相应的jar放到CLASSPATH中。在Hadoop1.X以前需要把hadoop-*-fairscheduler.jar从HADOOP_HOME/build/contrib/fairscheduler拷贝到HADOOP_HOME/lib，在Hadoop1.1.2版本在HADOOP_HOME/lib已经包含了公平调度器和能力调度器两个jar包

clip_image136

并需要在Hadoop的配置文件HADOOP_CONF_DIR/mapred-site.xml中设置下列属性让Hadoop启用公平调度器：

  <property>

    <name>mapred.jobtracker.taskScheduler</name>

    <value>org.apache.hadoop.mapred.FairScheduler</value>

  </property>

clip_image138

clip_image140

8.3.2验证启动公平调度器

重启集群

cd /usr/local/hadoop-1.1.2/bin

stop-all.sh

start-all.sh

clip_image142

通过jobtracker web UI查看：http://<JobTracker URL>/scheduler，在该实验地址为http://10.88.147.221:50030/scheduler，如下图所示

clip_image144

8.3.3准备数据

使用SSH工具（参见第1、2周2.1.3.1Linux文件传输工具所描述）把从NCDC下载的气象数据上传到上步骤创建的目录/usr/local/hadoop-1.1.2/input中。

clip_image016[1]

使用zcat命令把这些数据文件分别解压并合并到shortData.txt和longData.txt文件中，合并后把这两个文件上传到HDFS文件系统的/usr/hadoop/in目录中：

cd /usr/local/hadoop-1.1.2/input

zcat *.gz > shortData.txt

zcat *.gz > longData.txt

hadoop fs -copyFromLocal shortData.txt /usr/hadoop/in

hadoop fs -copyFromLocal longData.txt /usr/hadoop/in

hadoop fs -ls /usr/hadoop/in

clip_image146

clip_image148

8.3.4运行分析MapReduce

使用本周作业2打包的jar包，先启动MapReduceB，然后在另外一个登录Session中启动MapReduceA，执行输出目录为/usr/hadoop/out_long 和/usr/hadoop/out_short：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/longData.txt  /usr/hadoop/out_long

clip_image150

在另外一个登录Session运行情况：

cd /usr/local/hadoop-1.1.2

hadoop jar MinTemperature.jar MinTemperature /usr/hadoop/in/shortData.txt  /usr/hadoop/out_short

clip_image152

8.3.5观察结果

可以从上步骤结果图观察到：

MapReduceB 开始执行时间2014-10-14 16:51:13 结束时间：2014-10-14 16:53:36

MapReduceA 开始执行时间2014-10-14 16:51:20 结束时间：2014-10-14 16:53:28

MapReduceB这个job先运行，但因为这个job处理的数据量较大，而后一个MapReduceA的 job处理的数据量较小，所以MapReduceA的 job反而较早运行结束，可知公平调度器设置成功。

 

MapReduceB 作业ID为：job_201410141618_0009

MapReduceA 作业ID为：job_201410141618_0010

clip_image154

clip_image156

clip_image158

 

9    问题解决

9.1  在作业5中新增节点后，DataNode无法启动

通过拷贝Hadoop3（10.88.147.223）节点，修改名称后成为Hadoop4（10.88.147.224）节点，启动Hadoop3或者Hadoop4的DataNode会出现如下错误：

clip_image160

2014-10-15 16:28:40,344 WARN org.apache.hadoop.hdfs.server.datanode.DataNode: DataNode is shutting down: org.apache.hadoop.ipc.RemoteException: org.apache.hadoop.hdfs.protocol.UnregisteredDatanodeException: Data node 10.88.147.223:50010 is attempting to report storage ID DS-60030049-10.88.147.223-50010-1411456729315. Node 10.88.147.224:50010 is expected to serve this storage.

        at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getDatanode(FSNamesystem.java:4776)

        at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.processReport(FSNamesystem.java:3628)

        at org.apache.hadoop.hdfs.server.namenode.NameNode.blockReport(NameNode.java:1041)

        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)

        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

        at java.lang.reflect.Method.invoke(Method.java:606)

        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:578)

        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1393)

        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1389)

        at java.security.AccessController.doPrivileged(Native Method)

        at javax.security.auth.Subject.doAs(Subject.java:415)

        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1149)

        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:1387)

 

        at org.apache.hadoop.ipc.Client.call(Client.java:1107)

        at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:229)

        at com.sun.proxy.$Proxy5.blockReport(Unknown Source)

        at org.apache.hadoop.hdfs.server.datanode.DataNode.offerService(DataNode.java:1026)

        at org.apache.hadoop.hdfs.server.datanode.DataNode.run(DataNode.java:1527)

        at java.lang.Thread.run(Thread.java:745)

解决办法是修改Hadoop4节点中HADOOP_HOME/hdfs/data/current/VERSION文件，把storageID由DS-60030049-10.88.147.223-50010-1411456729315修改成DS-60030049-10.88.147.224-50010-1411456729315即可

clip_image162

 

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/8464.html

赞 (0)
0
上一篇 2021年7月18日
下一篇 2021年7月18日

相关推荐

发表回复

登录后才能评论