實驗40綜合實戰(zhàn)環(huán)境大數(shù)據(jù)

實驗四 要求實驗結(jié)束時，每位學(xué)生均已在服務(wù)器上運(yùn)行從2016年1月到6月這半24條環(huán)境數(shù)據(jù)，所有2500多個城市如果均如此進(jìn)行記錄，那每天產(chǎn)生的數(shù)據(jù)量將本實驗中選取了2016年1月到6月這半的每小時天氣和空氣質(zhì)量數(shù)據(jù)（未N/A~]#more在服務(wù)器上執(zhí)行下列命令，查看環(huán)境數(shù)據(jù)文件beijing.txt，路徑在~]#more40-140-1種可以看到，我們需要關(guān)心的數(shù)據(jù)有第一列DATEHOUR、第六列TMP和第七列AQI。將數(shù)據(jù)文件上傳至~]#hadoopfs-mkdir~]#hadoopfs-put~/data/40/beijing.txt在beijing.txtHDFS的在Eclipse上新建MapReduce項目，命名為TmpStat，在 下新建文importimportimportorg.apachehadoop.conf.Configuration;importorg.apachehadoopfs.Path;importorg.apachehadoop.io.IntWritable;importorg.apachehadoop.io.Text;importorg.apachehadoopmapreduce.Job;importorg.apachehadoopmapreduce.Mapper;importorg.apachehadoopimportorg.apachehadoopmapreduce.lib.input.TextInputFormat;importorg.apachehadoopmapreduce.lib.output.TextOutputFormat;importorg.apachehadoopmapreduce.lib.partition.HashPartitioner;publicclass{publicstaticclassStatMapperextendsMapper<Object,Text,Text,{privateIntWritableintValue=newIntWritable();privateTextdateKey=newText();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{String[]items=Stringdate=items[0];Stringtmp=dateKey.set(date.substring(0,6));context.write(dateKey,intValue);}}}{privateIntWritableresult=newpublicvoidreduce(Textkey,I ble<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{inttmpsum=0;intcount=0;for(IntWritableval:{tmp_sum+=val.get();}inttmp_avg=tmp_sum/count;result.set(tmpavg);context.write(key,result);}}publicstaticvoidmain(StringthrowsIOException,ClassNotFoundException,{Configurationconf=newJobjob=newJob(conf,"MonthlyAvgTmpStat");Jobjob=newJob(conf,"MonthlyAvgTmpStat");TextInputFormat.setInputPaths(job,args[0]); pletion(true)?0:1);}}使用Eclipse軟件將TmpStat項目導(dǎo)出成Jar文件，指定主類為TmpStat，命名為tmpstat.jar，并上傳至服務(wù)器上。~]#hadoopjartmpstat.jar/input/monthlyavgtmp 在上執(zhí)行tmpstat.jar~]#hadoopjartmpstat.jar/input/monthlyavgtmp 40-2 ~]#hadoopfs-ls~]#hadoopfs-cat/monthlyavgtmp/part-r-40-3查看月平均氣溫統(tǒng)計結(jié)在Eclipse上新建MapReduce項目，命名為AqiStatDaily，在 下新建文importimportimportorg.apachehadoop.conf.Configuration;importorg.apachehadoop.fs.Path;importorg.apachehadoop.io.IntWritable;importorg.apachehadoop.io.Text;importorg.apachehadoop.mapreduce.Job;importorg.apachehadoop.mapreduce.Mapper;importorg.apachehadoop.mapreduce.Reducer;importorg.apachehadoop.mapreduce.lib.input.TextInputFormat;importorg.apachehadoop.mapreduce.lib.partition.HashPartitioner;publicclass{publicstaticclassStatMapperextendsMapper<Object,Text,Text,{privateIntWritableintValue=newIntWritable();privateTextdateKey=newText();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{String[]items=Stringdate=items[0];Stringaqi=items[6];{context.write(dateKey,intValue);}}}publicstaticclassStatReducerextendsReducer<Text,IntWritable,Text,{privateIntWritableresult=newpublicvoidreduce(Textkey,I ble<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intaqi_sum=0;intcount=0;for(IntWritableval:{aqisum+=val.get();}intaqiavg=aqisum/count;context.write(key,result);}}publicstaticvoidmain(StringthrowsIOException,ClassNotFoundException,{Configurationconf=newConfiguration();Jobjob=newJob(conf,"AqiStatDaily");TextInputFormat.setInputPaths(job,args[0]);TextOutputFormat.setOutputPath(job,newPath(args[1])); pletion(true)?0:1);}}}使用Eclipse軟件將AqiStatDaily項目導(dǎo)出成Jar文件，指定主類為AqiStatDaily，命名為aqistatdaily.jar，并上傳至 ~]#hadoopjaraqistatdaily.jar/input/aqidaily 上執(zhí)行aqistatdaily.jar~]#hadoopjaraqistatdaily.jar/input/aqidaily 40-4在上查看統(tǒng)計結(jié)果文件。如圖40-5所示 ~]#hadoopfs-ls 40-5aqistatdaily.jar運(yùn)行結(jié)果文33~]#hadoopfs-cat 40-6~]#hadoopfs-cat/aqidaily/part-r-00000>~]#hadoopfs-cat/aqidaily/part-r-00001>>~]#hadoopfs-cat/aqidaily/part-r-00002>>~]#cataqidaily.txt|wc-在在HDFS上創(chuàng)建 ，并將aqidaily.txt上傳至 ~]#hadoopfs-mkdir~]#hadoopfs-putaqidaily.txt在Eclipse上新建MapReduce項目，命名為AqiStat，在 下新建文件importimportimportorg.apachehadoop.conf.Configuration;importorg.apachehadoopfs.Path;importorg.apachehadoop.io.IntWritable;importorg.apachehadoop.io.Text;importorg.apachehadoopmapreduce.Job;importorg.apachehadoopmapreduce.Mapper;importorg.apachehadoopimportorg.apachehadoopimportorg.apachehadoop.mapreduce.lib.output.TextOutputFormat;importorg.apachehadoop.mapreduce.lib.partition.HashPartitioner;publicclass{publicstaticfinalStringGOOD="優(yōu)";publicstaticfinalStringMODERATE良";publicstaticfinalStringLIGHTLY_POLLUTED="輕度污染";publicstaticfinalStringMODERA Y_POLLUTED="中度污染";publicstaticfinalStringHEAVILY_POLLUTED="重度污染";publicstaticfinalStringSEVERELY_POLLUTED="嚴(yán)重污染";publicstaticclassStatMapperextendsMapper<Object,Text,Text,{privatefinalstaticIntWritableone=newIntWritable(1);privateTextcond=newText();map方法，根據(jù)AQI值，將對應(yīng)空氣質(zhì)量的天數(shù)加publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{String[]items=value.toString().split("\t");intaqi=Integer.parseInt(items[1]);if(aqi<={//}elseif(aqi<={//}elseif(aqi<={//輕度污}elseif(aqi<={//中度污 Y}elseif(aqi<={//重度污}{//嚴(yán)重污}}}//reduce類，對相同的空氣質(zhì)量狀況，把它們<K,VList>中VList{privateIntWritableresult=newpublicvoidreduce(Textkey,I ble<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=for(IntWritableval:{sum+=}}}publicstaticvoidmain(StringthrowsIOException,ClassNotFoundException,{Configurationconf=newConfiguration();Jobjob=newJob(conf,"AqiStat");TextInputFo