MapReduce数据分析（10）综合实战

十、MapReduce综合实战

              综合实战：环境大数据

案列目的
1.学会分析环境数据文件；

2.学会编写解析环境数据文件并进行统计的代码；

3.学会进行递归MapReduce。

案例要求
要求实验结束时，每位学生均已在master服务器上运行从北京2016年1月到6月这半年间的历史天气和空气质量数据文件中分析出的环境统计结果，包含月平均气温、空气质量分布情况等。

实现原理

近年来，由于雾霾问题的持续发酵，越来越多的人开始关注城市相关的环境数据，包括空气质量数据、天气数据等等。
如果每小时记录一次城市的天气实况和空气质量实况信息，则每个城市每天都会产生24条环境数据，全国所有2500多个城市如果均如此进行记录，那每天产生的数据量将达到6万多条，每年则会产生2190万条记录，已经可以称得上环境大数据。
对于这些原始监测数据，我们可以根据时间的维度来进行统计，从而得出与该城市相关的日度及月度平均气温、空气质量优良及污染天数等等，从而为研究空气污染物扩散条件提供有力的数据支持。
本实验中选取了北京2016年1月到6月这半年间的每小时天气和空气质量数据（未取到数据的字段填充“N/A”），利用MapReduce来统计月度平均气温和半年内空气质量为优、良、轻度污染、中度污染、重度污染和严重污染的天数。

实验数据如下

第一题：编写月平均气温统计程序

import java.io.IOException;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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;public class TmpStat
{public static class StatMapper extends Mapper<Object, Text, Text, IntWritable>{private IntWritable intValue = new IntWritable();private Text dateKey = new Text();public void map(Object key, Text value, Context context)throws IOException, InterruptedException{String[] items = value.toString().split(",");String date = items[0];String tmp = items[5];if(!"DATE".equals(date) && !"N/A".equals(tmp)){//排除第一行说明以及未取到数据的行dateKey.set(date.substring(0, 6));intValue.set(Integer.parseInt(tmp));context.write(dateKey, intValue);}}}public static class StatReducer extends Reducer<Text, IntWritable, Text, IntWritable>{private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException{int tmp_sum = 0;int count = 0;for(IntWritable val : values){tmp_sum += val.get();count++;}int tmp_avg = tmp_sum/count;result.set(tmp_avg);context.write(key, result);}}public static void main(String args[])throws IOException, ClassNotFoundException, InterruptedException{Configuration conf = new Configuration();Job job = new Job(conf, "MonthlyAvgTmpStat");job.setInputFormatClass(TextInputFormat.class);TextInputFormat.setInputPaths(job, args[0]);job.setJarByClass(TmpStat.class);job.setMapperClass(StatMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setPartitionerClass(HashPartitioner.class);job.setReducerClass(StatReducer.class);job.setNumReduceTasks(Integer.parseInt(args[2]));job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);TextOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

运行结果：

第二题：编写每日空气质量统计程序

import java.io.IOException;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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;public class AqiStatDaily
{public static class StatMapper extends Mapper<Object, Text, Text, IntWritable>{private IntWritable intValue = new IntWritable();private Text dateKey = new Text();public void map(Object key, Text value, Context context)throws IOException, InterruptedException{String[] items = value.toString().split(",");String date = items[0];String aqi = items[6];if(!"DATE".equals(date) && !"N/A".equals(aqi)){dateKey.set(date);intValue.set(Integer.parseInt(aqi));context.write(dateKey, intValue);}}}public static class StatReducer extends Reducer<Text, IntWritable, Text, IntWritable>{private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException{int aqi_sum = 0;int count = 0;for(IntWritable val : values){aqi_sum += val.get();count++;}int aqi_avg = aqi_sum/count;result.set(aqi_avg);context.write(key, result);}}public static void main(String args[])throws IOException, ClassNotFoundException, InterruptedException{Configuration conf = new Configuration();Job job = new Job(conf, "AqiStatDaily");job.setInputFormatClass(TextInputFormat.class);TextInputFormat.setInputPaths(job, args[0]);job.setJarByClass(AqiStatDaily.class);job.setMapperClass(StatMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setPartitionerClass(HashPartitioner.class);job.setReducerClass(StatReducer.class);job.setNumReduceTasks(Integer.parseInt(args[2]));job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);TextOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

运行结果：

第三题：编写各空气质量天数统计程序

import java.io.IOException;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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;public class AqiStat
{public static final String GOOD = "优";public static final String MODERATE = "良";public static final String LIGHTLY_POLLUTED = "轻度污染";public static final String MODERATELY_POLLUTED = "中度污染";public static final String HEAVILY_POLLUTED = "重度污染";public static final String SEVERELY_POLLUTED = "严重污染";public static class StatMapper extends Mapper<Object, Text, Text, IntWritable>{private final static IntWritable one = new IntWritable(1);private Text cond = new Text();// map方法，根据AQI值，将对应空气质量的天数加1public void map(Object key, Text value, Context context)throws IOException, InterruptedException{String[] items = value.toString().split("\t");int aqi = Integer.parseInt(items[1]);if(aqi <= 50){// 优cond.set(GOOD);}else if(aqi <= 100){// 良cond.set(MODERATE);}else if(aqi <= 150){// 轻度污染cond.set(LIGHTLY_POLLUTED);}else if(aqi <= 200){// 中度污染cond.set(MODERATELY_POLLUTED);}else if(aqi <= 300){// 重度污染cond.set(HEAVILY_POLLUTED);}else{// 严重污染cond.set(SEVERELY_POLLUTED);}context.write(cond, one);}}// 定义reduce类，对相同的空气质量状况，把它们<K,VList>中VList值全部相加public static class StatReducer extends Reducer<Text, IntWritable, Text, IntWritable>{private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values,Context context)throws IOException, InterruptedException{int sum = 0;for (IntWritable val : values){sum += val.get();}result.set(sum);context.write(key, result);}}public static void main(String args[])throws IOException, ClassNotFoundException, InterruptedException{Configuration conf = new Configuration();Job job = new Job(conf, "AqiStat");job.setInputFormatClass(TextInputFormat.class);TextInputFormat.setInputPaths(job, args[0]);job.setJarByClass(AqiStat.class);job.setMapperClass(StatMapper.class);job.setCombinerClass(StatReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setPartitionerClass(HashPartitioner.class);job.setReducerClass(StatReducer.class);job.setNumReduceTasks(Integer.parseInt(args[2]));job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);TextOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

运行结果：

有什么不懂的，多多问问博主呦！！！

MapReduce数据分析（10）综合实战相关推荐

《Python数据分析与挖掘实战》第10章（下）——DNN2 筛选得“候选洗浴事件”3 构建模型
本文是基于<Python数据分析与挖掘实战>的实战部分的第10章的数据--<家用电器用户行为分析与事件识别> 做的分析. 接着前一篇文章的内容,本篇博文重点是处理用水事件中的属 ...
《Python数据分析与挖掘实战》示例源码免费下载
<Python数据分析与挖掘实战> 在当今大数据驱动的时代,要想从事机器学习.人工智能.数据挖掘等前沿技术,离不开数据跟踪与分析,通过NumPy.Pandas等进行数据科学计算,通过Se ...
23神经网络 :唐宇迪《python数据分析与机器学习实战》学习笔记
唐宇迪<python数据分析与机器学习实战>学习笔记 23神经网络 1.初识神经网络百度深度学习研究院的图,当数据规模较小时差异较小,但当数据规模较大时深度学习算法的效率明显增加,目前大 ...
小白都能学会的Python基础第六讲：综合实战2 - 大数据分词与词云图绘制
1.华小智系列 - Python基础(案例版) <Python基础>目录第六讲:综合实战2 - 大数据分词与词云图绘制 1.大数据分词技巧 2.词频统计技巧 3.词云图绘制 4.微博词云 ...
数据分析与挖掘实战-应用系统负载分析与磁盘容量预测
应用系统负载分析与磁盘容量预测背景某大型企业为了信息化发展的需要,建设了办公自动化系统.人力资源管理系统.财务管理系统.企业信息门户系统等几大企业级应用系统.因应用系统在日常运行时,会对底层软硬件 ...
工业数据分析技术与实战之运作优化——昆仑数据田春华培训听课记录
昆仑数据田春华老师在微信公众号的专栏培训:工业数据分析与实战.培训给出了一些实际的数据分析例子,包括"设备管理"."运作优化"和"营销服务" ...
工业数据分析技术与实战之设备管理——昆仑数据田春华培训听课记录
昆仑数据田春华老师在微信公众号的专栏培训:工业数据分析与实战.培训给出了一些实际的数据分析例子,包括"设备管理"."运作优化"和"营销服务" ...
《Python数据分析与挖掘实战》第15章 ——电商产品评论数据情感分析（LED）
文章目录 1.挖掘背景与目标 2.2 数据探索与预处理 2.1 数据筛选 2.2 数据去重 2.3 删除前缀评分 2.4 jieba分词 3 基于LDA 模型的主题分析 4.权重 5.如何在主题空间比 ...
《Python数据分析与挖掘实战》一第1章数据挖掘基础1.1　某知名连锁餐饮企业的困惑...
本节书摘来自华章出版社<Python数据分析与挖掘实战>一书中的第1章,第1.1节,作者张良均王路谭立云苏剑林,更多章节内容可以访问云栖社区"华章计算机"公众号 ...

MapReduce数据分析（10）综合实战

十、MapReduce综合实战

MapReduce数据分析（10）综合实战相关推荐

最新文章

热门文章