任务描述
知识点:
重? 点:
内? 容:
任务指导
(如在任务12中,按照手册已自行处理好2000-2022年的所有气象数据,也可跳过此步骤,使用自己处理好的数据文件即可,但需要在后续步骤中注意数据路径的问题)
先前按照任务12处理了2021-2022年数据,在后续气象预测部分任务需要2000-2022年的数据作为支持,所以现将处理后的(解压后)2000年-2022年的气象数据进行提供,可通过下述的URL下载地址进行下载
数据集路径:
格式:url/dataSet/systemLib/b3084be184684ee18f3b00b048bab0cc.zip,url参见实验窗口右侧菜单“实验资源下载”。
例如:https://staticfile.eec-cn.com/dataSet/systemLib/b3084be184684ee18f3b00b048bab0cc.zip
使用MapReduce对天气数据进行预处理,并在数据文件中添加对应基站ID,并将原来字段间的分隔符改为使用逗号分隔,以便于大Hive中使用该数据集。
当前在数据集中不包含基站编号字段,每个基站的编号体现在各个文件名的前5位,例如在“450010-99999-2000”文件中包含的是编号为“45001”的基站数据,所以需要将各个基站的编号添加到对应的数据文件中,并且在各个文件中每个字段之间的分隔符也是不一致的,所以也需要对数据进行清理,由于数据量较大,可以考虑使用MapReduce进行数据清理的工作。
基站编号 | 年 | 月 | 日 | 时间 | 温度 | 露点温度 | 气压 | 风向 | 风速 | 云量 | 1小时雨量 | 6小时雨量 |
---|---|---|---|---|---|---|---|---|---|---|---|---|
59997 | 2022 | 12 | 31 | 21 | 274 | 250 | 10133 | 70 | 20 | 5 | -9999 | -9999 |
任务实现
(如在任务12中,按照手册已自行处理好2000-2022年的所有气象数据,也可跳过此步骤,使用自己处理好的数据文件即可,但需要在后续步骤中注意数据路径的问题)
先前按照任务12处理了2021-2022年数据,在后续气象预测部分任务需要2000-2022年的数据作为支持,所以现将处理后的(解压后)2000年-2022年的气象数据进行提供,可通过下述的URL下载地址进行下载
数据集路径:
格式:url/dataSet/systemLib/b3084be184684ee18f3b00b048bab0cc.zip,url参见实验窗口右侧菜单“实验资源下载”。
例如:https://staticfile.eec-cn.com/dataSet/systemLib/b3084be184684ee18f3b00b048bab0cc.zip
# cd /home
# wget https://staticfile.eec-cn.com/dataSet/systemLib/b3084be184684ee18f3b00b048bab0cc.zip
# unzip /home/b3084be184684ee18f3b00b048bab0cc.zip
# hadoop fs -mkdir /china
# hadoop fs -put /home/china_data/* /china
天气的格式如下:
NCDC天气的格式说明:
气象要素包括:气温、气压、露点、风向风速、云量、降水量等。
年 | 月 | 日 | 时间 | 温度 | 露点温度 | 气压 | 风向 | 风速 | 云量 | 1小时雨量 | 6小时雨量 |
---|---|---|---|---|---|---|---|---|---|---|---|
2021 | 01 | 01 | 00 | 80 | -94 | 10285 | 50 | 60 | 1 | -9999 | -9999 |
当前在数据集中不包含基站编号字段,每个基站的编号体现在各个文件名的前5位,例如在“450010-99999-2000”文件中包含的是编号为“45001”的基站数据,所以需要将各个基站的编号添加到对应的数据文件中,并且在各个文件中每个字段之间的分隔符也是不一致的,所以也需要对数据进行清理,由于数据量较大,可以考虑使用MapReduce进行数据清理的工作。
使用MapReduce对天气数据进行ETL流程如下:
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.9.2</version>
</dependency>
</dependencies>
本次MapReduce任务的主要处理逻辑在Map函数中,在Map中获取当前正在处理的文件信息,通过文件信息获取相应的文件名,然后获取到文件名的前五位,前五位则是每个基站对应的基站编号,然后获取到数据文件中的每条数据并进行分割,分割后根据索引获取所需的数据,最后通过","对数据进行分隔,作为每个字段数据的新分隔符,根据所需重新将数据进行拼接
package com;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class ChinaMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 获取当前map正在处理的文件信息
InputSplit inputSplit = (InputSplit) context.getInputSplit();
// 获取文件名,例如:当前获取到“450010-99999-2000”
String fileName = inputSplit.toString().split("/")[5];
NullWritable val = NullWritable.get();
// 取出基站编号,例如:“45001”
String stn = fileName.substring(0,5);
// System.out.println(stn);
/** 获取所需字段
year=[] #年
month=[] #月
day=[] #日
hour=[] #时间
temp=[] #温度
dew_point_temp=[] #露点温度
pressure=[] #气压
wind_direction=[] #风向
wind_speed=[] #风速
clouds=[] #云量
precipitation_1=[] #1小时降水量
precipitation_6=[] #6小时降水量
*/
// 获取输入的每一条数据
String values = value.toString();
// 通过分隔符进行分割
String[] lines = values.split("\\s+");
String year = lines[0];
String month = lines[1];
String day = lines[2];
String hour = lines[3];
String temp = lines[4];
String dew_point_temp = lines[5];
String pressure = lines[6];
String wind_direction = lines[7];
String wind_speed = lines[8];
String cloud=lines[9];
String precipitation_1 = lines[10];
String precipitation_6 = lines[11];
// 使用“,”对每条数据进行拼接,每条数据的分隔符设置为","
String line = stn+","+year+","+month+","+day+","+hour+","+temp+","+dew_point_temp
+","+pressure+","+wind_direction+","+wind_speed+","+cloud+","+precipitation_1+","+precipitation_6;
System.out.println(line);
// 每条数据作为key进行输出
context.write(new Text(line),val);
}
}
package com;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class ChinaReducer extends Reducer<Text,NullWritable,Text,NullWritable> {
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
NullWritable val = NullWritable.get();
// 获取key
Text outLine = key;
context.write(outLine,val);
}
}
package com;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
public class ChinaDriver {
public static void main(String[] args) {
Configuration conf = new Configuration();
Job job = null;
try {
// 读取filename文件内容获取inputpath
BufferedReader br = new BufferedReader(new FileReader("C:\\installed\\filename.txt"));
String line = null;
ArrayList list = new ArrayList();
while((line=br.readLine())!=null){
list.add(line);
}
Path[] inputPath = new Path[list.size()];
for(int i = 0;i< inputPath.length;i++){
inputPath[i] = new Path(list.get(i).toString());
System.out.println(inputPath[i]);
}
job = Job.getInstance(conf);
job.setJarByClass(ChinaDriver.class);
job.setJobName("ChinaDriver");
// 设置Mapper类
job.setMapperClass(ChinaMapper.class);
// 设置Reducer类
job.setReducerClass(ChinaReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 设置输入路径
FileInputFormat.setInputPaths(job, inputPath);
// 设置输出路径
FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/china_all/"));
System.exit(job.waitForCompletion(true) ? 0 : 1);
} catch (IOException e) {
e.printStackTrace();
} catch (ClassNotFoundException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
由于本次MapRedcue任务需要处理2000-2022年的数据,每个年份的数据都保存在一个以年份命名的文件夹下,所以MapReduce需要读取22个文件夹下的所有数据,因此在运行MapReduce程序前,需要编写一个Shell脚本以生成MapReduce的inputPath文件,在该文件中包含所有需要处理的数据路径(该操作类似任务12中的generate_input_list.sh脚本)
# vim /home/shell/getHDFSfile.sh
#/bin/bash
rm -rf /home/filename.txt
# file = echo `hdfs dfs -ls /china | awk -F ' ' '{print $8}'`
for line in `hdfs dfs -ls /china | awk -F ' ' '{print $8}'`
do
filename="hdfs://master:9000$line"
echo -e "$filename" >> /home/filename.txt
done
# chmod u+x /home/shell/getHDFSfile.sh
# /home/shell/getHDFSfile.sh
# cat /home/filename.txt
进入master机器,查看运行结果最后5行数据:
# hadoop fs -cat /china_all/* | tail -5
数据格式说明:
基站编号 | 年 | 月 | 日 | 时间 | 温度 | 露点温度 | 气压 | 风向 | 风速 | 云量 | 1小时雨量 | 6小时雨量 |
---|---|---|---|---|---|---|---|---|---|---|---|---|
59997 | 2022 | 12 | 31 | 21 | 274 | 250 | 10133 | 70 | 20 | 5 | -9999 | -9999 |
上一个任务下一个任务