基于python的《Hadoop权威指南》一书中气象数据下载和map reduce化数据处理及其可视化

caodd1215

　　
　　文档内容：
　　　　1：下载《hadoop权威指南》中的气象数据
　　　　2：对下载的气象数据归档整理并读取数据
　　　　3：对气象数据进行map reduce进行处理
　　关键词：《Hadoop权威指南》气象数据　　map reduce　　python　　matplotlib可视化
　　
　　一：下载《hadoop权威指南》一书中的气象数据
　　《hadoop权威指南》一书中的气象数据位于 http://ftp3.ncdc.noaa.gov/pub/data/noaa/，
　　新建 getdata.py文件， 并加入如下代码：





1 #http://my.oschina.net/chyileon/blog/134915
2 import urllib
3 import urllib2
4 from bs4 import BeautifulSoup
5 import re
6 import os
7 import shutil
8
9 def getdata():
10     year = 1901
11     endYear = 1921
12     urlHead = 'http://ftp3.ncdc.noaa.gov/pub/data/noaa/'
13
14     while year < endYear:
15         if os.path.isdir(str(year)):
16             shutil.rmtree(str(year))
17         os.mkdir(str(year))
18
19         page = urllib2.urlopen(urlHead+str(year))
20         soup = BeautifulSoup(page, from_encoding="gb18030")
21
22         for link in soup.findAll('a'):
23             if link.getText().find('.gz') != -1:
24                 filename = link.getText()
25
26                 urllib.urlretrieve(urlHead+str(year)+'/'+filename, str(year)+'/'+filename)
27
28         year += 1
29
30 def main():
31     getdata()
32
33 if __name__=="__main__":
34     main()
View Code　　运行getdata.py,将在当前目录下生成数据文件
　　
　　二：对下载的气象数据归档整理并读取数据
　　说明：上一步骤在当前目录下生成【1901】~【1921】 共20文件，文件里是压缩的气象数据，本步骤知识将数据移动data文件夹下　
　　新建 movedata.py文件， 并加入如下代码：





1 import os
2 import shutil
3
4 def movedata():
5
6     curpath = os.getcwd()
7     list = os.listdir(curpath)
8     datapath = os.path.join(curpath, "data")
9     print(datapath)
10     for line in list:
11         filepath = os.path.join(curpath, line)
12         if os.path.isdir(filepath):
13             shutil.move(filepath,datapath)
14
15 def main():
16     movedata()
17
18 if __name__=="__main__":
19     main()
View Code 　　
　　三：对气象数据进行map reduce进行处理
　　说明：这里要读取文件中的数据内容，并通过将数据map reduce 化获取每年的最高、低温度
　　
　　1： 将文件中的数据内容逐行读出
　　新建reader.py文件，并加入如下代码：





1 import os
2 import gzip
3
4 def reader():
5
6     curpath = os.getcwd()
7     datapath = os.path.join(curpath, r"data")
8
9     for yearlist in os.listdir(datapath):
10         oneyearpath = os.path.join(datapath, yearlist)
11         datalist = os.listdir(oneyearpath)
12         for line in datalist:
13             onedatapath = os.path.join(oneyearpath, line)
14             with gzip.open(onedatapath, 'rb') as pf:
15                 print (pf.read())
16
17 def main():
18     reader()
19
20 if __name__=="__main__":
21     main()
View Code 　　测试上面代码：在命令行运行 reader.py，查看输出
　　2:通过mapper方法把数据处理成 "year \n temperature"的输出形式，如 "1901　　242",其中 242 表示温度为24.2度
　　新建mapper.py文件，并加入如下代码：　





1 import sys
2
3 def mapper(inlist):
4     for line in inlist:
5         if len(line) > 92:
6             year = (line[15:19])
7             if line[87] == '+':
8                 temperataure = line[88:92]
9             else:
10                 temperataure = line[87:92]
11             print year, temperataure
12
13 def main(inlist):
14     mapper(inlist)
15
16 if __name__=="__main__":
17     inlist = []
18     for line in sys.stdin:
19         inlist.append(line)
20     main(inlist)
View Code 　　测试上面代码：在命令行运行  reader.py | mapper.py ，查看输出。（注：这是是利用管道，把reader.py的输出作为mapper.py的输入）
　　3:通过reducer方法将mapper的输出数据整理并计算每年的最高、低温度，并输出
　　新建reducer.py文件，并加入如下代码：





1 import sys
2
3 def reducer(inlist):
4     cur_year = None
5     maxtemp = None
6     mintemp = None
7     for line in inlist:
8         year, temp = line.split()
9         try:
10             temp = int(temp)
11         except ValueError:
12             continue
13         if cur_year == year:
14             if temp > maxtemp:
15                 maxtemp = temp
16             if temp < mintemp:
17                 mintemp = temp
18         else:
19             if cur_year != None:
20                 print cur_year, maxtemp, mintemp
21             cur_year = year
22             maxtemp = temp
23             mintemp = temp
24     print cur_year, maxtemp, mintemp
25
26 def main(inlist):
27     reducer(inlist)
28
29 if __name__=="__main__":
30     inlist = []
31     for line in sys.stdin:
32         inlist.append(line)
33     main(inlist)
View Code　　　　测试上面代码：在命令行运行  reader.py | mapper.py | reducer.py，查看输出。
　　4：使用matplotlib对每年的最高、低数据进行可视化
　　新建drawer.py文件，并加入如下代码：





1 import sys
2 import matplotlib.pyplot as plt
3
4 def drawer(inlist):
5     yearlist = []
6     maxtemplist = []
7     mintemplist = []
8     for line in inlist:
9         year, maxtemp, mintemp = line.split()
10         try:
11             year = int(year)
12             maxtemp = int(maxtemp) / 10.
13             if(maxtemp) > 50:
14                 maxtemp = 50
15             mintemp = int(mintemp) / 10.
16         except ValueError:
17             continue
18         yearlist.append(year)
19         maxtemplist.append(maxtemp)
20         mintemplist.append(mintemp)
21     plt.plot(yearlist, maxtemplist, 'bd--')
22     plt.plot(yearlist, mintemplist, 'rp:')
23     plt.xlim(1901, 1920)
24     plt.ylim(-60, 80)
25     plt.title('min-max temperature for 1901-1920')
26     plt.xlabel('year')
27     plt.ylabel('temperature')
28     plt.legend(('max temp','min temp'), loc='upper right')
29     plt.show()
30     print(yearlist, maxtemplist, mintemplist)
31
32 def main(inlist):
33     drawer(inlist)
34
35 if __name__=="__main__":
36     inlist = []
37     for line in sys.stdin:
38         inlist.append(line)
39     main(inlist)
View Code 　　测试上面代码：在命令行运行  reader.py | mapper.py | reducer.py | drawer.py，查看输出。
　　显示效果如下图：（注：在前面处理的数据中， 可能由于采样的错误，会有出现999.9度的最高温度， 显然不符常理。在本例中，没有对此种错误进行深究，一致将超度50度的温度处理成50度）

　　
　　四 说明
　　1. 本例中，其实第二步 对下载的气象数据归档整理并读取数据 是多余的， 可以直接在第一步中修改文件存储目录跳过第二步。但为了熟悉python对文件的操作，还是将第二步的代码保留了下来。
　　2. 本例中，代码能运行得到实验目标，但并为对代码进行优化。请读者根据需要自行更改。
　　3. python代码的一大特点就是看起来像伪代码，又本例python代码比较简单，故没有给出注释。