现在的位置: 首页 > 综合 > 正文

Hadoop之气象站分析演示代码

2013年08月22日 ⁄ 综合 ⁄ 共 7609字 ⁄ 字号 评论关闭

一,背景

气象站分析一批复杂的数据,演示需要分析的数据

0043011990999991950051518004+68750+023550FM-12+038299999V0203201N00261220001CN9999999N9-00111+99999999999

存放在input.txt中。

其中包含了年份 和 温度数据

需要把这个年份和温度数据提取出来

 

二,具体执行

1,下载 hadoop-0.20.1

cd hadoop-020.1/conf/ 配置:

core-site.xml

Xml代码  收藏代码
  1. <?xml version="1.0"?>  
  2. <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>  
  3.   
  4. <!-- Put site-specific property overrides in this file. -->  
  5.   
  6. <configuration>  
  7.   <property>  
  8.     <name>fs.default.name</name>  
  9.     <value>hdfs://localhost:9000</value>  
  10.   </property>  
  11. </configuration>  

 hdfs-site.xml

Xml代码  收藏代码
  1. <?xml version="1.0"?>  
  2. <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>  
  3.   
  4. <!-- Put site-specific property overrides in this file. -->  
  5.   
  6. <configuration>  
  7.   <property>  
  8.     <name>dfs.replication</name>  
  9.     <value>1</value>  
  10.   </property>  
  11. </configuration>  

 mapred-site.xml

Xml代码  收藏代码
  1. <?xml version="1.0"?>  
  2. <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>  
  3.   
  4.     <!-- Put site-specific property overrides in this file. -->  
  5.   
  6. <configuration>  
  7.     <property>  
  8.         <name>mapred.job.tracker</name>  
  9.         <value>localhost:9001</value>  
  10.     </property>  
  11. </configuration>  

 配置完毕

 

cd bin

./hadoop namenode -format

./start-all.sh

 

2,我的pom.xml

Xml代码  收藏代码
  1. <?xml version="1.0" encoding="UTF-8"?>  
  2. <project  
  3.     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"  
  4.     xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">  
  5.     <modelVersion>4.0.0</modelVersion>  
  6.     <parent>  
  7.         <artifactId>balance</artifactId>  
  8.         <groupId>com.yajun</groupId>  
  9.         <version>1.0-SNAPSHOT</version>  
  10.     </parent>  
  11.     <groupId>com.yajun.hadoop</groupId>  
  12.     <artifactId>balance.hadoop</artifactId>  
  13.     <version>1.0-SNAPSHOT</version>  
  14.     <name>balance.hadoop</name>  
  15.     <url>http://maven.apache.org</url>  
  16.     <dependencies>  
  17.         <dependency>  
  18.             <groupId>junit</groupId>  
  19.             <artifactId>junit</artifactId>  
  20.             <version>4.7</version>  
  21.             <scope>test</scope>  
  22.         </dependency>  
  23.         <dependency>  
  24.             <groupId>org.mockito</groupId>  
  25.             <artifactId>mockito-core</artifactId>  
  26.             <version>1.8.2</version>  
  27.             <scope>test</scope>  
  28.         </dependency>  
  29.         <dependency>  
  30.             <groupId>org.apache.mahout.hadoop</groupId>  
  31.             <artifactId>hadoop-core</artifactId>  
  32.             <version>0.20.1</version>  
  33.         </dependency>  
  34.         <dependency>  
  35.             <groupId>commons-logging</groupId>  
  36.             <artifactId>commons-logging</artifactId>  
  37.             <version>1.1.1</version>  
  38.         </dependency>  
  39.         <dependency>  
  40.             <groupId>commons-httpclient</groupId>  
  41.             <artifactId>commons-httpclient</artifactId>  
  42.             <version>3.0</version>  
  43.         </dependency>  
  44.         <dependency>  
  45.             <groupId>commons-cli</groupId>  
  46.             <artifactId>commons-cli</artifactId>  
  47.             <version>1.2</version>  
  48.         </dependency>  
  49.     </dependencies>  
  50. </project>  

使用以上pom,用maven 构建eclipse开发环境

 

3,写代码

分析代码 (Map部分)

Java代码  收藏代码
  1. package com.yajun.hadoop.temperature;  
  2.   
  3. import java.io.IOException;  
  4.   
  5. import org.apache.hadoop.io.IntWritable;  
  6. import org.apache.hadoop.io.LongWritable;  
  7. import org.apache.hadoop.io.Text;  
  8. import org.apache.hadoop.mapred.MapReduceBase;  
  9. import org.apache.hadoop.mapred.Mapper;  
  10. import org.apache.hadoop.mapred.OutputCollector;  
  11. import org.apache.hadoop.mapred.Reporter;  
  12.   
  13. /** 
  14.  * hadoop 书上的例子,提取年份,温度数据 
  15.  *  
  16.  * @author txy 
  17.  */  
  18. public class MaxTemperatureMapper extends MapReduceBase implements  
  19.         Mapper<LongWritable, Text, Text, IntWritable> {  
  20.   
  21.     public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output,  
  22.                     Reporter reporter) throws IOException {  
  23.         String line = value.toString();  
  24.         // 提取年份  
  25.         String year = line.substring(1519);  
  26.         // 提取温度   
  27.         String temp = line.substring(8792);  
  28.         if (!missing(temp)) {  
  29.             int airTemperature = Integer.parseInt(temp);  
  30.             output.collect(new Text(year), new IntWritable(airTemperature));  
  31.         }  
  32.     }  
  33.   
  34.     /** 
  35.      * 如果提取出来的温度达到9999,认为是提取不到数据 
  36.      *  
  37.      * @param temp 
  38.      * @return 是否能正确提取温度数据 
  39.      */  
  40.     private boolean missing(String temp) {  
  41.         return temp.equals("+9999");  
  42.     }  
  43.   
  44. }  

 

输出结果代码(Reduce部分)

Java代码  收藏代码
  1. package com.yajun.hadoop.temperature;  
  2.   
  3. import java.io.IOException;  
  4. import java.util.Iterator;  
  5.   
  6. import org.apache.hadoop.io.IntWritable;  
  7. import org.apache.hadoop.io.Text;  
  8. import org.apache.hadoop.mapred.MapReduceBase;  
  9. import org.apache.hadoop.mapred.OutputCollector;  
  10. import org.apache.hadoop.mapred.Reducer;  
  11. import org.apache.hadoop.mapred.Reporter;  
  12.   
  13. /** 
  14.  * 输出当年最高温度 
  15.  *  
  16.  * @author txy 
  17.  */  
  18. public class MaxTemperatureReducer extends MapReduceBase implements  
  19.         Reducer<Text, IntWritable, Text, IntWritable> {  
  20.     public void reduce(Text key, Iterator<IntWritable> values,  
  21.                        OutputCollector<Text, IntWritable> output, Reporter reporter)  
  22.             throws IOException {  
  23.         int maxValue = Integer.MIN_VALUE;  
  24.         while (values.hasNext()) {  
  25.             maxValue = Math.max(maxValue, values.next().get());  
  26.         }  
  27.         output.collect(key, new IntWritable(maxValue));  
  28.     }  
  29. }  

 

运行整个JOB的代码

Java代码  收藏代码
  1. package com.yajun.hadoop.temperature;  
  2.   
  3. import org.apache.hadoop.conf.Configured;  
  4. import org.apache.hadoop.fs.Path;  
  5. import org.apache.hadoop.io.IntWritable;  
  6. import org.apache.hadoop.io.Text;  
  7. import org.apache.hadoop.mapred.FileInputFormat;  
  8. import org.apache.hadoop.mapred.FileOutputFormat;  
  9. import org.apache.hadoop.mapred.JobClient;  
  10. import org.apache.hadoop.mapred.JobConf;  
  11. import org.apache.hadoop.util.Tool;  
  12. import org.apache.hadoop.util.ToolRunner;  
  13.   
  14. public class MaxTemperatureDriver extends Configured implements Tool {  
  15.     @Override  
  16.     public int run(String[] args) throws Exception {  
  17.         if (args.length != 2) {  
  18.             System.err.printf("Usage: %s [generic options] <input> <output>\n", getClass()  
  19.                     .getSimpleName());  
  20.             ToolRunner.printGenericCommandUsage(System.err);  
  21.             return -1;  
  22.         }  
  23.         JobConf conf = new JobConf(getConf(), getClass());  
  24.         conf.setJobName("Max temperature");  
  25.         FileInputFormat.addInputPath(conf, new Path(args[0]));  
  26.         FileOutputFormat.setOutputPath(conf, new Path(args[1]));  
  27.         conf.setOutputKeyClass(Text.class);  
  28.         conf.setOutputValueClass(IntWritable.class);  
  29.         conf.setMapperClass(MaxTemperatureMapper.class);  
  30.         conf.setCombinerClass(MaxTemperatureReducer.class);  
  31.         conf.setReducerClass(MaxTemperatureReducer.class);  
  32.         JobClient.runJob(conf);  
  33.         return 0;  
  34.     }  
  35.   
  36.     public static void main(String[] args) throws Exception {  
  37.         int exitCode = ToolRunner.run(new MaxTemperatureDriver(), args);  
  38.         System.exit(exitCode);  
  39.     }  
  40. }  

 

4,eclipse环境的hadoop插件配置好(如果没有安装这个插件也很简单:https://issues.apache.org/jira/browse/MAPREDUCE-1262 上面下载,扔到eclipse 的dropins目录里面搞定)
与hadoop的配置一样

 

5,运行代码

现将input.txt拷贝到 hdfs中去

 

./hadoop fs -put /home/txy/work/balanceofworld/balance/balance.hadoop/src/main/resources/temperature/input.txt /user/txy/src/main/resources/temperature/input.txt

 

设置运行 MaxTemperatureDriver 的时候需要两个命令行参数

1,输入文件:src/main/resources/temperature/input.txt (对应到HDFS里面的

/user/txy/src/main/resources/temperature/input.txt)

2,输出文件:src/main/resources/temperature/output.txt(对应到HDFS里面的

/user/txy/src/main/resources/temperature/output.txt

 

然后就在eclipse 里右键在hadoop上运行吧,哈哈。

该博文转载自http://yjhexy.iteye.com/blog/608105

抱歉!评论已关闭.