距離上次發佈資料研究文章好一陣子，主要在把過去走馬看花的 Hadoop 再複習，並學習了一些相關生態圈的內容，Hadoop 生態圈己經發展了相當完整，列舉其中幾個項目，希望今年能夠熟悉他們並能應用在實際的資料上：
- 基礎: Hadoop, HBase, ZooKeeper
- 資料處理相關: Hive, Pig, Chukwa
- 綜合及與ML相關: Mahout, Spark
- DevOps: Ansible, Puppet，其實不一定要使用工具，Shell Script 也可以達成
回到這次的應用，PM2.5這個指標近幾年在台灣的新聞開始被重視起來，且環保署也作了一個相當不錯的即時監測的網頁－行政院環境保護署－空氣品質監測網，但作資料分析就是喜歡追根究底一下其中的細節及統計，就來看看全年中較顯著的地方。
** PM2.5全年統計 **
我的目標是希望能在每天的0-23小時，針對個別被測量的最高的值站別累積計數一次，會得到：
日期，站別，小時，最大計數(或最小計數)
這樣的日計數統計，並且能按小時來統計 PM 2.5 最高值與最低值的各站別。因為各小時的全年統計量在這表達上有些困難，故在這只列出全年24小時最高前 10 名與最低量計數的前 20 名。
- 全年量測最高計數 Top 10
| 觀測站 | 每小時最高量計數加總 |
|---|---|
| 埔里 | 704.0 |
| 馬祖 | 596.0 |
| 金門 | 507.0 |
| 林園 | 484.0 |
| 竹山 | 457.0 |
| 崙背 | 395.0 |
| 潮州 | 388.0 |
| 復興 | 316.0 |
| 斗六 | 301.0 |
| 小港 | 295.0 |
住在這些觀測站附近的人可能要隨時準備好保護措施。在這前 3 名蠻令人意外，好山好水的埔里竟然是第一名的壞空氣，我們還常去那裡渡假！另外 Top 2,3 的金門馬祖，依據新聞的說詞是來自東北的沙塵及工業污染，以及廈門污染物。
- 全年量測最低計數 Top 20
| 觀測站 | 每小時最低量計數加總 |
|---|---|
| 前鎮 | 1314.0 |
| 恆春 | 1311.0 |
| 花蓮 | 1210.0 |
| 陽明 | 1176.0 |
| 菜寮 | 884.0 |
| 湖口 | 860.0 |
| 楠梓 | 768.0 |
| 臺東 | 751.0 |
| 三重 | 748.0 |
| 淡水 | 628.0 |
| 沙鹿 | 564.0 |
| 竹東 | 553.0 |
| 古亭 | 550.0 |
| 宜蘭 | 550.0 |
| 麥寮 | 524.0 |
| 橋頭 | 464.0 |
| 林園 | 403.0 |
| 仁武 | 380.0 |
| 關山 | 370.0 |
| 豐原 | 368.0 |
這個 Top 1 的資料有待再確認，因為我印象中前鎮是工業區，Google 了相關資料，似乎沒那麼樂觀，可能要回頭看看觀測站的資料。
更新：前鎮觀測站的資料有蠻多時段為0，這真的有待澄清是己改善或人為疏忽。或是以其他統計量來估計。
** 資料來源及內容 **
- 資料來源: 行政院環境保護署－歷年監測資料下載, 可惜的目前(2017/01)僅提供 2015 年度資料
- 資料格式：試算表檔案(.xls或.ods)，這裡就比較頭疼，所以我希望能快速處理 79 個檔案，借助了 xls2csv 轉換 csv 以利後續載入 Hive，這裡有個坑，如果在 Hive 的 Schema 裡沒特別指定字串的界限符，最好先把雙引號處理掉，否則在 Hive 查詢時會誤認自己的 HQL 寫錯
- 資料的格式為: 日期，觀測站, 觀測項, 00,01,02, 03,04,05,06,07,08,09,10,11,12,13,14,15,16,17,18,19,20,21,22,23，其中我們只取觀測項為 PM2.5 的資料，其實有為數可觀的測量項目，日後都可以拿來探索一番
** 資訊技術 **
這次使用的資訊技術較為多一些，大致整理3個方面：
- Hadoop: 1 name node + 2 data nodes
- Hive: 使用 mysql 為 Metastore
- Python: pyhive 連接 Hive， Pandas 統計資料
處理流為 Python <-> PyHive <-> Hive <-> Haddop
** 資訊處理 **
- 建立 Hive Table
create table TWOO (rdate date, station string, mitem string, r0 float,r1 float,r2 float,r3 float,r4 float,r5 float,r6 float,r7 float,r8 float,r9 float,r10 float,r11 float,r12 float,r13 float,r14 float,r15 float,r16 float,r17 float,r18 float,r19 float,r20 float,r21 float,r22 float,r23 float)
row format delimited fields terminated by ','
lines terminated by '\n'
tblproperties("skip.header.line.count"="1");
其中 skip.header.line.count 是為了忽略 csv 檔內的第一行標題欄。
- 測試 HQL，HQL 跟 SQL 還是有一些差異，而且都必須轉換為 Hadoop 的 Map-Reduce 程式，並等待各節點的計算，所以為了減少程式的測試時間，先完成 HQL。這裡有個難題，因為是按日期及按小時去取得最大值，如果按天x按小時這樣的個別擊破方式得到最高或最低觀測站，會是相當可觀的處理量(24 x 365=8760)，2個data node的處理量，或採並行處理的方式，都是相當粍時的，且不符合 ** Hadoop 大規模讀取的特性**。還好 Hive 2 以後的版本有支援 aggregation 的處理：
rank() over (PARTITION BY rdate ORDER BY r0 DESC)
- Python 上場了，記得先把 Hive Server 開啟。這裡我原本調用Pyhs2，碰到了不少問題，也有人說Pyhs2己經不再維護，後來改採用PyHive來作連接，然後再一一把r0~r23 組合成 HQL 交由 Hive 處理。這裡取得最高值的一個循環(r0-r23)，使用系統的量測時間為：
real 26m53.756s
user 0m0.999s
sys 0m0.043s
其實這時間處理有些長，應該試看看把 13,086,511 資料弄到 MySQL 來比較處理的時間。最後轉換為 Pandas 的 Data Frame 就由強大的Pandas，一行即得統計結果：
dd.loc[dd['MaxHourCount']>0, \
['Station','MaxHourCount']].groupby(['Station']).sum().sort('MaxHourCount',ascending=False).head(20)
最後來看看 Hive 轉換為 Hadoop Map/Reduce 的成果, _________
