首页 > 代码库 > Hadoop第5周练习—MapReduce计算气象温度等例子

Hadoop第5周练习—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    书面作业5MapReduce是否可以自动识别新增节点?... 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.2G6G内存

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

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

l  JDK1.7.0_55 64 bit

l  Hadoop1.1.2

1.2  机器网络环境

集群包含三个节点:1namenode2datanode,其中节点之间可以相互ping通。节点IP地址和主机名分布如下:

序号

IP地址

机器名

类型

用户名

运行进程

1

10.88.147.221

hadoop1

名称节点

hadoop

NNSNNJobTracer

2

10.88.147.222

hadoop2

数据节点

hadoop

DNTaskTracer

3

10.88.147.223

hadoop3

数据节点

hadoop

DNTaskTracer

所有节点均是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  PaaSPaaS(Platform-as-a- Service):平台即服务。PaaS实际上是指将软件研发的平台作为一种服务,以SaaS的模式提交给用户。因此,PaaS也是SaaS模式的一种应用。但是,PaaS的出现可以加快SaaS的发展,尤其是加快SaaS应用的开发速度。例如:软件的个性化定制开发。

l  SaaSSaaS(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 = http://www.mamicode.com/Integer.MAX_VALUE;

        for(IntWritable value : values) {

            minValue = http://www.mamicode.com/Math.min(minValue, value.get());

        }

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

    }

}

3.3  实现过程

3.3.1编写代码

进入/usr/local/hadoop-1.1.2/myclass目录,在该目录中建立MinTemperature.javaMinTemperatureMapper.javaMinTemperatureReducer.java代码文件,代码内容为3.2所示,执行命令如下:

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

vi MinTemperature.java

vi MinTemperatureMapper.java

vi MinTemperatureReducer.java

技术分享

MinTemperature.java

技术分享

MinTemperatureMapper.java

技术分享

MinTemperatureReducer.java:

技术分享

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

技术分享

3.3.3打包编译文件

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

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

ls

mv *.jar ..

rm *.class

技术分享

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                

技术分享

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

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

技术分享

使用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

技术分享

技术分享

气象数据具体的下载地址为 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

技术分享

3.3.7查看结果

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

hadoop fs -ls /usr/hadoop/out

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

技术分享

3.3.8通过页面结果

1.     查看jobtracker.jsp

http://10.88.147.221:50030/jobtracker.jsp

技术分享

已经完成的作业任务:

技术分享

任务的详细信息:

技术分享

技术分享

2.     查看dfshealth.jsp

http://10.88.147.221:50070/dfshealth.jsp

技术分享

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

技术分享

技术分享

 

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 = http://www.mamicode.com/0;

        long numValue = http://www.mamicode.com/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 = http://www.mamicode.com/0;

        long numValue = http://www.mamicode.com/0;

        int avgValue = http://www.mamicode.com/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.javaAvgTemperatureMapper.javaAvgTemperatureCombiner.javaAvgTemperatureReducer.java代码文件,代码内容为4.3所示,执行命令如下:

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

vi AvgTemperature.java

vi AvgTemperatureMapper.java

vi AvgTemperatureCombiner.java

vi AvgTemperatureReducer.java

技术分享

AvgTemperature.java

技术分享

AvgTemperatureMapper.java

技术分享

AvgTemperatureCombiner.java

技术分享

AvgTemperatureReducer.java:

技术分享

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

技术分享

4.4.3打包编译文件

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

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

ls

mv *.jar ..

rm *.class

技术分享

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

技术分享

技术分享

4.4.5查看结果

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

hadoop fs -ls /usr/hadoop/out1

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

技术分享

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.shreducer.sh代码文件,代码内容为5.2所示,执行命令如下:

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

vi mapper.sh

vi reducer.sh

技术分享

mapper.sh

技术分享

reducer.sh

技术分享

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

技术分享

5.3.3查看结果

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

hadoop fs -ls /usr/hadoop/out_awk

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

技术分享

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.pyreducer.py代码文件,代码内容为6.2所示,执行命令如下:

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

vi mapper.py

vi reducer.py

技术分享

mapper.py

技术分享

reducer.py

技术分享

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

技术分享

6.3.3查看结果

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

hadoop fs -ls /usr/hadoop/out_py

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

技术分享

 

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

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

DNTaskTracer

7.3.1环境准备

参考第三周第二题第四个问题--SNNNN分离--进行环境准备,增加一个新节点并确保各节点能够免密码SSH访问

7.3.1.1复制虚拟机

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

技术分享

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

设置该节点IP地址为:10.88.147.224

技术分享

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

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

sudo vi /etc/sysconfig/network

技术分享

技术分享

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

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

sudo vi /etc/hosts

技术分享

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

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

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

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

技术分享

7.3.1.6重启所有虚拟机

技术分享

7.3.1.7配置ssh免密码登录

1.     hadoop410.88.147.244)节点中使用ssh-keygen -t rsa生成私钥和公钥;

技术分享

2.     hadoop410.88.147.244)节点中公钥信息加入到authorized_keys文件中;

chmod 400 -R /home/hadoop/.ssh

cat id_rsa.pub >> authorized_keys

cat authorized_keys

技术分享

3.     authorized_keys分发到各个节点上;

scp authorized_keys hadoop@hadoop1:/home/hadoop/.ssh

技术分享

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

技术分享

7.3.2准备数据

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

技术分享

先把这三年的数据放到/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

技术分享

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

hadoop fs -ls /usr/hadoop/in

技术分享

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

技术分享

查看运行作业节点信息,Hadoop集群由两个DataNode节点,分别为Hadoop2Hadoop3

技术分享

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

技术分享

运行作业后,启动新增节点Hadoop4中的DataNodeTaskTracker进程

cd /usr/local/hadoop-1.1.2/bin

hadoop-daemon.sh start datanode

hadoop-daemon.sh start tasktracker

jps

技术分享

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

技术分享

技术分享

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

技术分享

 

7.3.5重启Hadoop集群,运行任务

停止Hadoop集群运行:

cd /usr/local/hadoop-1.1.2/bin

stop-all.sh

技术分享

Hadoop4节点中结束DataNodeTaskTracker进程:

cd /usr/local/hadoop-1.1.2/bin

hadoop-daemon.sh stop datanode

hadoop-daemon.sh stop tasktracker

技术分享

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

cd /usr/local/hadoop-1.1.2/conf

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

技术分享

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

cd /usr/local/hadoop-1.1.2/bin

start-all.sh

技术分享

技术分享

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

cd /usr/local/hadoop-1.1.2

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

技术分享

技术分享

 

7.3.6结论

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

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

8.1  书面作业6内容

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

8.2  程序代码

参见作业2代码并打成jar

8.3  实现过程

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

8.3.1打开公平调度器

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

技术分享

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

  <property>

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

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

  </property>

技术分享

技术分享

8.3.2验证启动公平调度器

重启集群

cd /usr/local/hadoop-1.1.2/bin

stop-all.sh

start-all.sh

技术分享

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

技术分享

8.3.3准备数据

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

技术分享

使用zcat命令把这些数据文件分别解压并合并到shortData.txtlongData.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

技术分享

技术分享

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

技术分享

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

cd /usr/local/hadoop-1.1.2

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

技术分享

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处理的数据量较大,而后一个MapReduceAjob处理的数据量较小,所以MapReduceAjob反而较早运行结束,可知公平调度器设置成功。

 

MapReduceB 作业ID为:job_201410141618_0009

MapReduceA 作业ID为:job_201410141618_0010

技术分享

技术分享

技术分享

 

9    问题解决

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

通过拷贝Hadoop310.88.147.223)节点,修改名称后成为Hadoop410.88.147.224)节点,启动Hadoop3或者Hadoop4DataNode会出现如下错误:

技术分享

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文件,把storageIDDS-60030049-10.88.147.223-50010-1411456729315修改成DS-60030049-10.88.147.224-50010-1411456729315即可

技术分享

 

Hadoop第5周练习—MapReduce计算气象温度等例子