使用Hadoop分析气象数据（附代码）
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
刚学了Hadoop在网上找完整的分析使用过程进行练手观看。本文数据和方法均来自于大佬的使用Hadoop分析气象数据完整版附带完整代码侵删
文章目录
注意为了不出现横向拖拉命令里我加了换行所有命令都是如此。复制进命令行时记得去掉换行
1.获取数据
1.1下载数据
wget -D --accept-regex=REGEX -P data -r
-c ftp://ftp.ncdc.noaa.gov/pub/data/noaa/isd-lite/2019/5*
下载时可能会出现卡顿中断下载按Ctrl+c结束下载然后重复下载就可以了网不好的多试几次
我是全部下载完了我这时候一共有这么多
1.2 数据格式
截取部分数据格式如下
2019 01 01 00 -65 -123 10199 345 95 8 -9999 -9999
2019 01 01 01 -62 -115 10213 350 86 -9999 -9999 -9999
2019 01 01 02 -62 -110 10223 343 86 -9999 -9999 -9999
2019 01 01 03 -62 -114 10234 337 77 -9999 -9999 -9999
2019 01 01 04 -62 -118 10242 345 86 -9999 -9999 -9999
2019 01 01 05 -62 -116 10252 331 63 -9999 -9999 -9999
2019 01 01 06 -62 -114 10259 306 38 6 -9999 -9999
2019 01 01 07 -62 -114 10264 281 29 -9999 -9999 -9999
2019 01 01 08 -62 -113 10268 268 39 -9999 -9999 -9999
2019 01 01 09 -59 -116 10271 254 31 3 -9999 -9999
2019 01 01 10 -62 -115 10271 238 24 -9999 -9999 -9999
2019 01 01 11 -80 -122 10269 254 12 -9999 -9999 -9999
2019 01 01 12 -67 -103 10264 322 12 5 -9999 -9999
2019 01 01 13 -62 -100 10261 27 13 -9999 -9999 -9999
2019 01 01 14 -29 -72 10259 230 40 -9999 -9999 -9999
2019 01 01 15 -20 -67 10254 242 49 5 -9999 -9999
字段解释如下
字段1位置1-4长度4:观测年份四舍五入到最接近
字段2位置6-7长度2观察月四舍五入到最接近
字段3位置9-11长度2观察日四舍五入到最接近
字段4位置12-13长度2观察时四舍五入到最接近
字段5位置14-19长度6空气温度单位摄氏度比例因子10缺少值-9999
字段6位置20-24长度6露点温度为了达到饱和必须在恒定的压力和水蒸气含量下
冷却给定的空气包的温度。单位摄氏度比例因子10缺少值-9999
字段7:Pos 26-31长度6海平面压力相对于平均海平面的气压。单位公顷比例因
子10缺少值-9999
字段832-37号位置长度6风向。正北角在正北和风向之间以顺时针方向测量的角度。
单位角度。比例因子1缺少值-9999。*注静风风向编码为0。
字段938-43位置长度6风速空气通过一个固定点的水平运动速度。单位米每秒。
比例因子10。缺少值-9999
字段10:位置44-49长度6:天空状况总覆盖代码表示被云层或其它遮蔽现象覆盖的整个穹
顶的一部分的代码。缺少值-9999
域
0:无SKC或CLR
1 一个okta-1/10或更小但不是零
2 两个oktas-2/10-3/10或几个
3 三个oktas-4/10
4 四个oktas-5/10或SCT
5 五个oktas-6/10
6 六个oktas-7/10-8/10
7 七个oktas-9/10或以上但不是10/10或BKN
8 八个oktas-10/10或OVC
9 天空模糊不清或云量无法估计
10 部分遮蔽
11 稀散
12 分散的
13 暗散射
14 薄断
15 破碎的
16 暗断
17 薄阴
18 阴天
19 阴天
字段11位置50-55长度6液体沉淀深度尺寸-持续一小时在一个小时的积累期内测量
的液体沉淀的深度。单位毫米比例因子10缺少值-9999。*注痕量降水编码为-1
字段12位置56-61长度6液体沉淀深度尺寸-持续6小时在六小时的积累期内测量的液
体沉淀的深度。单位毫米。比例因子10。缺少值-9999。*注痕量降水编码为-1
字段很多但是这里我只用前5个字段任务是统计**每日最高温度、最低温度、平均温度**有时间的话顺便计算点儿静态统计值。其余字段应该是类似的正所谓一通百通。
1.3 合并数据
数据很分散合并数据将2019目录下的所有以.gz结尾的的文件合并到data.txt
zcat 2019/*.gz > data.txt
到此数据获取完毕
2.MapReduce处理数据
2.1 环境配置启动集群
可以查看原作者的文章 大数据学习系列Hadoop3.0苦命学习一本文不再赘述。
2.2 文件上传到HDFS
要保证hadoop已经启动
hdfs dfs -mkdir -p /usr/hadoop/in
hdfs dfs -ls /usr/hadoop/
hdfs dfs -put data.txt /usr/hadoop/in/
查看是否上传成功
hdfs dfs -ls /usr/hadoop/in
2.3 编写MapReduce代码
2.3.1 TemperatureMapper
public class TemperatureMapper extends Mapper<LongWritable, Text,
Text, LongWritable> {
private static final long MISSING = -9999;
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
Iterable<String> split = Splitter.on(" ").omitEmptyStrings().split(line);
ArrayList<String> arrayList = new ArrayList<>(16);
for (String s : split) {
arrayList.add(s);
}
// 过滤掉字段不足的数据
if (arrayList.size() >= 5) {
String month = arrayList.get(1);
String day = arrayList.get(2);
long temperature = Long.parseLong(arrayList.get(4));
// 过滤掉温度不存在的数据
if (Math.abs(temperature - MISSING) > 0.0001) {
context.write(new Text(month + "/" + day),
new LongWritable((temperature)));
}
}
}
}
主要是原数据进行了清洗过滤了一些不合格的数据。
2.3.2 TemperatureReducer
public class TemperatureReducer extends Reducer<Text, LongWritable,
Text, Temperature> {
@Override
protected void reduce(Text key, Iterable<LongWritable> values,
Context context) throws IOException, InterruptedException {
long maxTemperature = Long.MIN_VALUE;
long minTemperature = Long.MAX_VALUE;
double avgTemperature = 0.0;
long temp;
int count = 0;
if (values!=null) {
for (LongWritable value: values) {
temp = value.get();
maxTemperature = Math.max(temp, maxTemperature);
minTemperature = Math.min(temp, minTemperature);
avgTemperature += temp;
count++;
}
Temperature temperature = new Temperature(maxTemperature,
minTemperature, avgTemperature/count);
context.write(key, temperature);
}
}
}
计算出每日温度的最大值、最小值和平均值并放入Temperature对象中。
2.3.3 JobMain
public class JobMain extends Configured implements Tool {
@Override
public int run(String[] strings) throws Exception {
// 创建一个任务对象
Job job = Job.getInstance(super.getConf(), "mapreduce_temperature");
// 打包放在集群运行时需要做一个配置
job.setJarByClass(JobMain.class);
// 第一步设置读取文件的类K1和V1
job.setInputFormatClass(TextInputFormat.class);
TextInputFormat.addInputPath(job,
new Path("hdfs://node01:8020/usr/hadoop/in"));
// 第二步设置Mapper类
job.setMapperClass(TemperatureMapper.class);
// 设置Map阶段的输出类型k2和v2的类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
// 第三、四、五、六步采用默认方式分区排序规约分组
// 第七步设置Reducer类
job.setReducerClass(TemperatureReducer.class);
// 设置Reduce阶段的输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Temperature.class);
// 第八步设置输出类
job.setOutputFormatClass(TextOutputFormat.class);
// 设置输出路径
TextOutputFormat.setOutputPath(job,
new Path("hdfs://node01:8020/usr/hadoop/temperature"));
boolean b = job.waitForCompletion(true);
return b?0:1;
}
public static void main(String[] args) throws Exception {
Configuration configuration = new Configuration();
// 启动一个任务
ToolRunner.run(configuration, new JobMain(), args);
}
}
2.4 执行程序
2.4.1 打包上传
可以百度一下
2.4.2 运行
hadoop jar temperature_test-1.0-SNAPSHOT.jar cn.sky.hadoop.JobMain
temperature_test-1.0-SNAPSHOT.jar 为打包后的jar包名
cn.sky.hadoop.JobMain 主程序的目录名
下面为我分析后的数据部分截图。因为我们每个人取得不一样所以这部分也会一样出来结果就行
3.导入数据到Hive
关于Hive的介绍可以看原作者的博文大数据学习系列Hadoop3.0苦命学习五
有个问题若直接从HDFS导入数据到HiveHDFS上的数据会丢失。
所以我将数据下载下来重命名为 ```temperature_data ``
数据有了开始创建Hive表
create external table temperature (t_date string, t_max double,
t_min double, t_avg double) row format delimited fields terminated by '\t';
加载数据到hive
load data local inpath '/export/services/temperature_data' overwrite
into table temperature;
查前面5条数据看一眼
select * from temperature limit 5;
4.Hive数据分析
弄得简单就查几个静态数据吧。
4.1 查询2019全年平均温度
select avg(t_avg) from temperature;
最终结果是68.77因为数据是被放大了10倍左右
4.2 查询2019全年高于平均温度的天数
select count(1) from temperature where t_avg > 68.7;
答案是198天
结束演示查询别的数据可以再自己写相关的语句
5.使用Sqoop导入数据到Mysql
关于Sqoop的相关介绍可以查看原作者的博文大数据学习系列Hadoop3.0苦命学习七
在使用前保证已经安装好MySQL
5.1Mysql创建数据库
CREATE TABLE `temperature` (
`Tem_Date` varchar(10) NOT NULL,
`Tem_Max` double DEFAULT NULL,
`Tem_Min` double DEFAULT NULL,
`Tem_Avg` double DEFAULT NULL,
PRIMARY KEY (`Tem_Date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
5.2 开始导入
bin/sqoop export --connect jdbc:mysql://192.168.0.102:3306/userdb
--username root --password 123456 --table temperature
--export-dir /usr/hadoop/temperature --input-fields-terminated-by "\t"
可以去MySQL查看一下自己的数据。
但是这里我自己操作时没有使用Sqoop成功。也没有使用Linux环境下的数据库。所以是将数据导入到WIndows中来了
6.展示数据
这里前端使用echartjsp后端使用Spring、SpringMVC、Mybatis。
代码较多展示主要的。
6.1 前端代码
主要是这一段使用Ajax向后台请求数据然后将数据丢进eChart中。
6.2 后端代码
controller层
@Autowired
private TemperatureService tempService;
@RequestMapping("/getTemperature.action")
@ResponseBody
public TemperatureReturnPojo getTemperature(){
TemperatureReturnPojo temperaturePojo = tempService.getAllTemperature();
System.out.println(temperaturePojo);
return temperaturePojo;
}
Service层
public interface TemperatureService {
TemperatureReturnPojo getAllTemperature();
}
Service实现类
@Service
public class TemperatureServiceImpl implements TemperatureService {
@Autowired
private TemperatureMapper temperatureMapper;
@Override
public TemperatureReturnPojo getAllTemperature() {
TemperatureReturnPojo temperatureReturnPojo
= new TemperatureReturnPojo();
ArrayList<String> dates = new ArrayList<>();
ArrayList<String> maxs = new ArrayList<>();
ArrayList<String> mins = new ArrayList<>();
ArrayList<String> avgs = new ArrayList<>();
DecimalFormat df = new DecimalFormat("#.00");
List<TemperaturePojo> allTemperature
= temperatureMapper.getAllTemperature();
for (TemperaturePojo pojo : allTemperature) {
dates.add(pojo.getTem_Date());
maxs.add(df.format(pojo.getTem_Max()/10.0));
mins.add(df.format(pojo.getTem_Min()/10.0));
avgs.add(df.format(pojo.getTem_Avg()/10.0));
}
temperatureReturnPojo.setTem_Dates(dates);
temperatureReturnPojo.setTem_Maxs(maxs);
temperatureReturnPojo.setTem_Mins(mins);
temperatureReturnPojo.setTem_Avgs(avgs);
return temperatureReturnPojo;
}
}
实体类
public class TemperaturePojo {
private String Tem_Date;
private Double Tem_Max;
private Double Tem_Min;
private Double Tem_Avg;
// 省略Get()、Set()、ToString()方法
}
public class TemperatureReturnPojo {
private List<String> Tem_Dates;
private List<String> Tem_Maxs;
private List<String> Tem_Mins;
private List<String> Tem_Avgs;
// 省略Get()、Set()、ToString()方法
}
Mapper
public interface TemperatureMapper {
List<TemperaturePojo> getAllTemperature();
}
<mapper namespace="cn.itcast.weblog.mapper.TemperatureMapper" >
<select id="getAllTemperature"
resultType="cn.itcast.weblog.pojo.TemperaturePojo">
select * from temperature;
</select>
</mapper>
页面运行结果展示
流程完成撒花~~~
原作者的代码链接GitHub链接–Hadoop-Analyzes-Weather-Data
问题解决
原作者的前端代码的这个Index.jsp文件可以直接替换到外边那个index就可以直接启动Tomcat后访问到了
