開始著手分析數(shù)據(jù)渡处，數(shù)據(jù)是導(dǎo)出成csv格式的文本文件，每一行格式如下：
1701170830490656,"2014-01-06 22:31:18.608000",8.64,1
各個(gè)列的含義依次是：地點(diǎn)，發(fā)生時(shí)間戳，發(fā)生重量，發(fā)生類型锯茄。我們需要按地點(diǎn)對(duì)發(fā)生重量進(jìn)行統(tǒng)計(jì)排序輸出，按發(fā)生類型對(duì)發(fā)生重量進(jìn)行統(tǒng)計(jì)排序輸出茶没。
由于集群上運(yùn)行時(shí)需要每個(gè)工作節(jié)點(diǎn)都要能訪問到文件肌幽，所以在此次實(shí)踐中，hdfs還沒有搭建成功之前抓半，讀取本地文件分析只能用spark的local模式喂急，重點(diǎn)時(shí)spark rdd的初步入門。
首先運(yùn)行
./bin/spark-shell --master=local
建議先在ide中寫好代碼笛求，再粘貼到shell中運(yùn)行
type TeamWeightEntity =? (String, Float)
type TypeWeightEntity =? (Int,Float)
以上代碼：定義了兩個(gè)元組廊移，分別用來存放（地點(diǎn)，發(fā)生重量）和（類型探入，發(fā)生重量）狡孔。
def parseLine(line:String):TeamWeightEntity= {
? val a = line.split(",")
? (a(0), a(2).toFloat)
}
def? parseLineByType(line:String):TypeWeightEntity= {
? val a = line.split(",")
? (a(3).toInt, a(2).toFloat)
}
以上代碼：定義了兩個(gè)每行記錄的解析函數(shù)。
val filePath ="/home/zhujianfeng/team_mw.csv"
val linesRDD = sc.textFile(filePath)
以上代碼：指明了數(shù)據(jù)文件路徑蜂嗽，并加載到spark中形成RDD苗膝。
我在mac上實(shí)踐時(shí)，路徑稍有不同植旧，執(zhí)行結(jié)果如同這樣：
接下來辱揭，我們對(duì)這個(gè)rdd進(jìn)行處理：
val weightRDD = linesRDD.map(parseLine).cache()
以上代碼，我們得到一個(gè)存儲(chǔ)了地點(diǎn)病附、發(fā)生重量的RDD问窃，并將其緩存起來。
val sumWeight = weightRDD.reduceByKey((a,b)=>a+b).sortBy(e=>e._2,false)
以上代碼做了兩個(gè)操作胖喳，首先是根據(jù)key(也就是地點(diǎn))進(jìn)行reduce，相同的key的操作是值相加贮竟。接著對(duì)產(chǎn)生的中間結(jié)果進(jìn)行排序丽焊，sortBy兩個(gè)參數(shù)较剃，一個(gè)是排序依據(jù)的函數(shù)，一個(gè)是表示倒序的布爾值技健。
最后組織輸出：
sumWeight.collect()
同樣的方式我們可以再做一次写穴，形成類型統(tǒng)計(jì)的結(jié)果：
val typeWeight = linesRDD.map(parseLineByType).cache()
val sumWeight2 = typeWeight.reduceByKey((a,b)=>a+b).sortBy(e=>e._2,false)
sumWeight2.collect()
接下來使用spark sql方式分析數(shù)據(jù)，在此之前先將文件放到hdfs中雌贱。首先要定義讀取存儲(chǔ)數(shù)據(jù)記錄的類啊送。
case class TeamWeight(tid:String,recTs:String,mwWeight:Float,mwType:Int)
接著讀取文件，處理并形成DataFrame欣孤。
val filePath ="hdfs://Y40/medical_waste/team_mw.csv"
import spark.sqlContext.implicits._
val teamDF = sc.textFile(filePath).map(_.split(",")).map(attributes =>TeamWeight(attributes(0), attributes(1), attributes(2).toFloat, attributes(3).toInt)).toDF()
teamDF.createOrReplaceTempView("t_team_weight")
以上代碼import是必須的馋没，否則RDD會(huì)沒有toDF()方法。邏輯很簡(jiǎn)單降传，分析每行篷朵，創(chuàng)建TeamWeight
對(duì)象數(shù)組的RDD，然后轉(zhuǎn)成DataFrame婆排。
val df1 = spark.sql("SELECT tid,SUM(mwWeight) AS sumWeight FROM t_team_weight GROUP BY tid ORDER BY sumWeight DESC")
df1.show()
以上代碼就是執(zhí)行sql語(yǔ)句声旺，并展示查詢結(jié)果。
同樣段只，類型的匯總排行查代碼如下：
val df2 = spark.sql("SELECT mwType,SUM(mwWeight) AS sumWeight FROM t_team_weight GROUP BY mwType ORDER BY sumWeight DESC")
df2.show()
最后腮猖，我們實(shí)踐一下部署過程，將代碼編譯后大成jar包赞枕，注意因?yàn)闆]用到spark之外的庫(kù)澈缺，所以只要把代碼打包就行了，IDEA的設(shè)置如下圖：
將打包好的jar文件上傳到集群中的任何一臺(tái)機(jī)器鹦赎，然后運(yùn)行spark-commit：
./bin/spark-submit --master spark://Y40:7077 /home/zhujianfeng/medical_waste_spark.jar
由于這個(gè)jar打包時(shí)已經(jīng)指定了main class谍椅，所以不需要再加--class參數(shù)，如果需要執(zhí)行另一個(gè)類古话，就要加參數(shù)指定類名了雏吭。
