你好,游客 登录 注册 搜索
背景:
阅读新闻

Hadoop - 国内各站点最高温度、气压和风速统计

[日期:2015-01-20] 来源:博客园  作者:mikevictor [字体: ]

一、简介

    该实例统计国内各个站点的最高温度(为节省篇幅只以温度为例,可稍作修改即可统计气压与风速),数据来源于汇总在NCDC的天气气球数据集中(包含世界大量数据集,该实例只分析国内站点,数据对外公开,可下载)。

二、数据准备与预处理

    从NCDC下载的天气气球数据集(ftp://ftp.ncdc.noaa.gov/pub/data/igra或http: //www1.ncdc.noaa.gov/pub/data/igra/ , 压缩为gz包)如下,可见并不适合Hadoop的MR模块处理,需要进行 预处理(本例下载数据gz包总大小为293MB,解压缩后为1.43GB):

复制代码
#5928719630901009999   5
10 85000  1481B  202B-9999  190   20
10 70000  3139B  142B-9999  180   20
10 50000  5880B  -55B-9999   60   30
10 40000  7605B -142B-9999  100   40
10 30000  9750B -255B-9999  100   70
#5928719630901129999   7
10 85000  1481B  215B-9999  320   20
10 70000  3142B  132B-9999  300   10
10 50000  5889B  -35B-9999   50   30
10 40000  7620B -125B-9999  100   40
10 30000  9759B -275B-9999   70   60
10 20000 12561B -482B-9999   90  110
10 10000 16788B -785B-9999   90  100
复制代码

首先需要阅读下载相关目录的readme.txt,才能站点相关字段的含义,温度数据已经*10(为了保留一位小数)

以该数据为例(其中的 9999一般代表数据缺失):

#5052719630901009999 5
10 85000 1314B 68B-9999-9999-9999

数据头部
标识 站点编号 日期 观察起始时间 观察结束时间 记录数
# 50527 19630901 00 9999 5

 

数据记录          
Major Level Minor Level 气压(Pa)3-8 气压标识 位势高度(米)10-14 位势高度标识 温度*10(16- 20位) 温度标识 露点下降 风力方向 风速(m/s)
1 0 85000 空格 1314 B 68 B -9999 -9999 -9999

 由于MapReduce一行行处理数据,而该数据记录部分依赖于数据头部,MR对数据进行分区时对它们分开的可能性非常大,所以每条数据记录部分必须加上头部的部分信息(根据需要确定),即预处理,对数据预处理的结果可输出到本地,然后再拷贝至HDFS。

    在本例中,数据头部只关注<站点编号>、<日期>,数据头部与数据记录形成的新格式如下:

预处理后的数据格式              
505271963090110 85000  1314B  68B-9999-9999-9999                
站点编号 日期 Major Level Minor Level 气压(Pa) 气压标识 位势高度(米)10-14位 位势高度标识 温度*10 温度标识 露点下降 风力方向 风速(m/s)
50527 19630901 1 0 85000 空格 1314 B 68 B -9999 -9999 -9999

即如下面格式的新格式:

592871963090110 85000  1481B  202B-9999  190   20
592871963090110 70000  3139B  142B-9999  180   20
592871963090110 50000  5880B  -55B-9999   60   30

 

三、数据拷贝至HDFS

    数据从本地拷贝至HDFS可以通过编码,也可使用eclipse的hadoop插件进行(该插件目前一般需要根据自己的环境编译得到jar放入eclipse的plugins文件夹下,过程稍微繁琐),

当然也可以使用bin/hadoop工具copyFromLocal进行(不过需要先复制到集群中的任意一台机器),本例中把数据存放在HDFS的 /weatherballoon 目录下,以下代码可供参考:

 

复制代码
core-site.xml:不同的配置文件方便本地测试和集群切换,在MR程序调试的时候很有用

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
     <property>
         <name>fs.default.name</name>
         <value>hdfs://hbase-01:9000</value>
     </property>
</configuration>
复制代码
复制代码
public class CopyFromLocalMain {
    
    private static Configuration config = new Configuration();
    
    public static void main(String[] args) throws Exception {
        config.addResource("core-site.xml");
        File inputDir = new File("d:/weatherballoon/input/");
        String hdfsDir = "/weatherballoon/input/";
        
        if (!inputDir.exists()) {
            System.err.println(inputDir.getAbsolutePath() + " directory not exist .");
            System.exit(-1);
        }
        File[] files = inputDir.listFiles();
        if (files == null) {
            System.err.println(inputDir.getAbsolutePath() + " directory is empty .");
            System.exit(-1);
        }
        
        for (File file : files) {
            copyFileToHdfs(file, hdfsDir + file.getName());
        }
        System.out.println("Copy finished, total: " + files.length);
    }
public static void copyFileToHdfs(File local,String dest) throws IOException{ InputStream in = new BufferedInputStream(new FileInputStream(local)); FileSystem fs = FileSystem.get(config); FSDataOutputStream out = fs.create(new Path(dest)); IOUtils.copyBytes(in, out, 4096, true); out.close(); } }
复制代码

数据拷贝完毕后可访问HDFS的namenode查看状态(默认50070端口),本例状态如下图:

 

 

四、编写MapReduce程序

    目前的数据格式已经每行之间无依赖性,首先编辑Mapper部分,该部分用于把站点ID作为key的数据集存入OutputCollector中:

复制代码
public class MaxTemperatureMapper 
    extends MapReduceBase implements Mapper<LongWritable, Text, Text, RecordValue>{
    
    public static final int DATA_LENGTH = 49; //预处理后的数据行长度
    
    @Override
    public void map(LongWritable key, Text value,
            OutputCollector<Text, RecordValue> output, Reporter reporter) throws IOException {
        
        //505271963090110 85000  1314B   68B-9999-9999-9999
        String line = value.toString();
        if (line.length() != DATA_LENGTH) {
            System.out.println("------------->Error record : " + line);
            return;
        }
        
        String stationId = line.substring(0, 5);
        String date = line.substring(5, 13);
        String temp = line.substring(28, 33);
        if (!missing(temp)) {
            int temperature = Integer.parseInt(temp.trim());
            output.collect(new Text(stationId), new RecordValue(date, temperature));
        }
    }
    
    private boolean missing(String temp) {
        return temp.equals("-9999");
    }
    
}
复制代码

    Mapper中输出的Value值为自定义的类型(即RecordValue),因为需要同时记录日期和温度,如果要自定义类型,则必须实现Writable(如Hadoop自带的LongWritable,IntWritable,Text等)。

    实现该接口使得对象能够序列化在不同机器间传输(进程间采用RPC通信,Hadoop采用Avro序列化,其他比较流行的序列化框架有Apache Thrift和Google protocol buffers),

一般建议实现WritableComparable接口,该接口中有个compareTo 方法的实现对于MapReduce来说是比较重要的,用于基于键的中间结果排序。

也可以实现RawComparator ,即可在字节流中排序,而不需要反序列化,减小额外开销。

RecordValue的实现如下:

复制代码
public class RecordValue implements WritableComparable<RecordValue>{
    
    private String date;
    private int temperature;
    
    public RecordValue(){}
    public RecordValue(String date, int temperature) {
        this.date = date;
        this.temperature = temperature;
    }
    @Override
    public void write(DataOutput out) throws IOException {
        out.write(date.getBytes());
        out.writeInt(temperature);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
        byte[] buff = new byte[8];
        in.readFully(buff);
        date = new String(buff);
        temperature = in.readInt();
    }
    @Override
    public int compareTo(RecordValue o) {
        if (date.compareTo(o.getDate()) > 0 || temperature > o.getTemperature()) {
            return 1;
        }
        return -1;
    }
    @Override
    public String toString() {
        return " -- " + date + "," + temperature;
    }
    
    //省略setter getter
}
复制代码

Mapper部分需要做单元测试,成功后接下面编写Reducer部分:

复制代码
public class MaxTemperatureReducer extends MapReduceBase implements
        Reducer<Text, RecordValue, Text, RecordValue> {

    @Override
    public void reduce(Text key, Iterator<RecordValue> values,
            OutputCollector<Text, RecordValue> output, Reporter reporter)
            throws IOException {
        
        int maxValue = Integer.MIN_VALUE;
        String date = "00000000";
        while (values.hasNext()) {
            RecordValue record = values.next();
            int temp = record.getTemperature();
            if (temp > maxValue) {
                maxValue = temp;
                date = record.getDate();
            }
        }
        output.collect(key, new RecordValue(date, maxValue));
    }
    
}
复制代码

当Reduce部分单元测试成功后即可编写驱动程序MaxTemperatureDriver,先用本地小数据集进行测试,配置文件切换为本地配置,如:

复制代码
public static void main(String[] args) throws IOException {
        
        String inputPath = "file:///d:/weatherballoon/input/*";
        String outputPath = "file:///d:/weatherballoon/output";
        
        File out = new File("d:/weatherballoon/output");
        if (out.exists()) {
            FileUtils.forceDelete(out); //采用apache common io包
        }
        
        Configuration conf = new Configuration();
        conf.addResource("core-site-local.xml");
        
        JobConf job = new JobConf(conf);
        job.setJobName("Max Temperature(NCDC)");
        job.setMapperClass(MaxTemperatureMapper.class);
        job.setReducerClass(MaxTemperatureReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(RecordValue.class);
        
        FileInputFormat.addInputPath(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        JobClient.runJob(job);
    }
复制代码

运行程序,如果出错则在本地容易查出错误地方,查看输出结构是否如何预期,下面是本例小部分数据集的输出结果:

50527 -- 20040721,358
50557 -- 20100627,342
50603 -- 19730409,440

温度已经乘以10,故对应的结果如下表格:

站点ID 日期 温度(摄氏度)
50527 20040721 35.8
50557 20100627 34.2
50603 19730409 44.0

五、集群运行

测试成功后可切换至集群运行,更改MaxTemperatureDriver的main,如下:

复制代码
public static void main(String[] args) throws IOException {
        
        String inputPath = "/weatherballoon/input/";
        String outputPath = "/weatherballoon/output";
        
        Configuration conf = new Configuration();
        
        JobConf job = new JobConf(conf);
        job.setJobName("Max Temperature(NCDC)");
        job.setMapperClass(MaxTemperatureMapper.class);
        job.setReducerClass(MaxTemperatureReducer.class);
job.setJarByClass(MaxTemperatureDriver.class); //!important job.setOutputKeyClass(Text.
class); job.setOutputValueClass(RecordValue.class); FileInputFormat.addInputPath(job, new Path(inputPath)); FileOutputFormat.setOutputPath(job, new Path(outputPath)); JobClient.runJob(job); }
复制代码

 

然后程序打包(编写MANIFEST.MF):

Manifest-Version: 1.0
Class-Path: .
Main-Class: org.mike.hadoop.weatherballoon.MaxTemperatureDriver

eclipse->export->jar并选择MANIFEST.MF文件,把jar上传到集群任一节点,执行如下命令:

hadoop jar MaxTemperature.jar

 

运行如下图:

 

成功后即可从HDFS拷贝结果文件至本地查看(或者直接hadoop dfs -cat也可以),本例得到的结果如下(列出小部分):

 

复制代码
50527    -- 20040721,358
50557    -- 20100627,342
50603    -- 19730409,440
50745    -- 19990413,386
50774    -- 20100624,316
50834    -- 19920428,426
50953    -- 20010604,346
51076    -- 19931031,506
51133    -- 19870716,600
51156    -- 19860309,552
51232    -- 19800802,220
复制代码

根据NCDN中igra-stations.txt文件得到对应的站点整理后如下:

站点ID 站点名称 日期 最高温度
50527 HAILAR 20040721 35.8
50557 NENJIANG 20100627 34.2
50603 CHIN-BARAG 19730409 44
50745 CHICHIHAR 19990413 38.6
50774 YICHUN 20100624 31.6
50834 TA KO TAI 19920428 42.6
50953 HARBIN 20010604 34.6
51076 ALTAY 19931031 50.6
51133 TA CHENG 19870716 60
51156 HOBOG SAIR 19860309 55.2
51232 BORDER STATION 19800802 22

 

Finished ..

本文章版权归本人及博客园共同所有,转载请标明原文出处(http://www.cnblogs.com/mikevictor07/),以下内容为个人理解,仅供参考。





收藏 推荐 打印 | 录入:Cstor | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款