AQI.py.zip_AQI_hadoop_python_空气质量指数

#coding=utf-8 import pymysql con = pymysql.connect( host = 'localhost', port = 3306, user = 'root', passwd = 'root', db = 'lqioc_ioc_yw', charset = 'utf8') cur = con.cursor() #连接数据库 pollutant3 = {'SO2':'SO2_24H','NO2':'NO2_24H','PM10':'PM10','O3-1H':'O3_1H','PM2.5':'PM2.5','CO(mg/m3)':'CO_24H','O3-8H':'O3_8H'} pollutant = {'O3-8H':'O3_8H'} pollutant2 = {'SO2':'SO2_1H','NO2':'NO2_1H','PM10':'PM10','O3-1H':'O3_1H','PM2.5':'PM2.5','CO(mg/m3)':'CO_1H'} error_date = ['0.0(E)','-99.0(E)',None] MONITOR_CODE = ['H01','H02','H03'] #找到时间 def get_day_id1(): sql9 = "SELECT DISTINCT day_id from h_monitor_day" cur.execute(sql9) result8 = cur.fetchall() return result8 def get_day_id2(): sql9 = "SELECT DISTINCT day_id from h_monitor_hour" cur.execute(sql9) result7 = cur.fetchall() return result7 def cal_aqi_24H(): for x in range(len(MONITOR_CODE)): MONITOR_CODE1 = MONITOR_CODE[x] day_id1 = get_day_id2() for i in range(len(day_id1)): day_id2 = day_id1[i][0] for m in pollutant: # print(m, str(pollutant[m])) sql1 = "select `%s`,TIME,MONITOR_CODE from h_monitor_hour where DAY_ID ='%s' and MONITOR_CODE='%s';" % (m, day_id2,MONITOR_CODE1) cur.execute(sql1) result1 = cur.fetchall() list_A_B = [] for i in range(len(result1)): print(result1[i][0]) if result1[i][0] not in error_date: sql2 = "select MIN(a.Cl1)as Cl,MAX(a.Ch1)as Ch from ( select (case WHEN `%s` > %s then `%s` else null end) as Cl1,(case WHEN `%s` < %s then `%s` else null end) as Ch1 from h_pollutant_limit) a;" \ % (str(pollutant[m]), result1[i][0], str(pollutant[m]), str(pollutant[m]),result1[i][0], str(pollutant[m])) print(sql2) cur.execute(sql2) result2 = cur.fetchall() if result2[0][0] == None: sql4 = "insert into hour_tmp(C,TIME,PRI,Cl,Ch,Il,Ih,AQI) values(%s,'%s','%s',0,%s,%s,%s,500;" \ % (result1[i][0], result1[i][1], m, result2[0][1], result3[0][0], result3[0][1]) print(sql4) text_save(list_A_B, 'cal_aqi_24H.sql') else: sql3 = "select MIN(b.Il1)as Il,MAX(b.Ih1)as Ih from (select (case when `%s`=%s then IAQI else null end ) as Il1,(case when `%s`=%s then IAQI else null end ) as Ih1 from h_pollutant_limit) b;" \ % (str(pollutant[m]), result2[0][0], str(pollutant[m]), result2[0][1]) print(sql3) cur.execute(sql3) result3 = cur.fetchall() sql4 = "insert into hour_tmp(C,TIME,PRI,Cl,Ch,Il,Ih,AQI,MONITOR_CODE) values(%s,'%s','%s',%s,%s,%s,%s,(Ih-Il)/(Ch-Cl)*(C-Cl)+Il,'%s');" \ % (result1[i][0], result1[i][1], m, result2[0][0], result2[0][1],result3[0][0], result3[0][1],MONITOR_CODE1) list_A_B.append(sql4) print(sql4) text_save(list_A_B, 'cal_aqi_24H.sql') # cur.execute(sql4) # con.commit() print('cal_aqi_24H执行成功！') def cal_aqi_1H(): for x in range(len(MONITOR_CODE)): MONITOR_CODE1 = MONITOR_CODE[x] day_id1 = get_day_id1() for i in range(len(day_id1)): day_id2 = day_id1[i][0] for m in pollutant: # print(m, str(pollutant[m])) sql5 = "select `%s`,TIME from h_monitor_day where DAY_ID ='%s' and MONITOR_CODE='%s';" % (m, day_id2, MONITOR_CODE1) cur.execute(sql5) result4 = cur.fetchall() list_C_D = [] for i in range(len(result4)): print(result4[i][0]) if result4[i][0] not in error_date: sql6 = "select MIN(a.Cl1)as Cl,MAX(a.Ch1)as Ch from ( select (case WHEN `%s` > %s then `%s` else null end) as Cl1,(case WHEN `%s` < %s then `%s` else null end) as Ch1 from h_pollutant_limit) a;" \ % (str(pollutant[m]), result4[i][0], str(pollutant[m]), str(pollutant[m]), result4[i][0],str(pollutant[m])) # print(sql6) cur.execute(sql6) result5 = cur.fetchall() if result5[0][0] == None: sql8 = "insert into day_tmp(C,TIME,PRI,Cl,Ch,Il,Ih,AQI) values(%s,'%s','%s',0,%s,%s,%s,500;" \ % (result4[i][0], result4[i][1], m, result5[0][1], result6[0][0], result6[0][1]) print(sql8) text_save(list_C_D, 'cal_aqi_1H.sql') else: sql7 = "select MIN(b.Il1)as Il,MAX(b.Ih1)as Ih from (select (case when `%s`=%s then IAQI else null end ) as Il1,(case when `%s`=%s then IAQI else null end ) as Ih1 from h_pollutant_limit) b;" \ % (str(pollutant[m]), result5[0][0], str(pollutant[m]), result5[0][1]) # print(sql7) cur.execute(sql7) result6 = cur.fetchall() sql8 = "insert into day_tmp(C,TIME,PRI,Cl,Ch,Il,Ih,AQI,MONITOR_CODE) values(%s,'%s','%s',%s,%s,%s,%s,(Ih-Il)/(Ch-Cl)*(C-Cl)+Il,'%s');" \ % (result4[i][0], result4[i][1], m, result5[0][0], result5[0][1], result6[0][0],result6[0][1],MONITOR_CODE1) print(sql8) list_C_D.append(sql8) text_save(list_C_D, 'cal_aqi_1H.sql') #将读取的数据存入txt文档中 def text_save(content,filename,mode='a'): #打开文件 file = open(filename,mode) for i in range(len(content)): number = content[i] + '\n' file.write(number) file.close() if __name__ == '__main__': get_day_id2() cal_aqi_24H() get_day_id1() cal_aqi_1H() # SELECT B.TIME,B.aqi2,A.PRI FROM tmp2 A right JOIN (select TIME,MAX(AQI) as aqi2 FROM TMP2 GROUP BY TIME) B ON A.TIME=B.TIME AND A.AQI = B.aqi2; #目标表需要更新字段：AOI,AOI_LEV,PRIMARY,AOI_TYPE # 关联字段：TIME #http://www.kaixian.tv/gd/2015/0220/12682851.html 数值超过上限的解决办法，aqi直接为500最大值！