//groupbykey
一、准備數據
val flights=sc.textFile("data/Flights/flights.csv")
val sampleFlights=sc.parallelize(flights.take(1000))
val header=sampleFlights.first
val filteredFlights=sampleFlights.filter(line=>{
line!=header&&line.split(",")(22)!=""
})
這里的准備數據使用的相對路勁
二、使用map函數獲得自己想要計算的幾個字段
val airLinesMap=filteredFlights.map(line=>{
val tailNum=line.split(",")(6)
val airline=line.split(",")(4)
(airline,tailNum)
})
三、使用groupbykey操作，合並行
val airlinesGroup= airLinesMap.distinct.groupByKey()
airlinesGroup.take(20).foreach(println)
四、計算每個航空公司的航班，當然也可以不用groupbykey直接使用reducebykey實現
//計算每個航空公司的航班
val airplanesCount =airlinesGroup.map(line=>{
(line._1,line._2.size)
})
airplanesCount.take(20).foreach(println)
五。計算飛機延誤的幾率
//計算延誤的幾率
val flightsMap=filteredFlights.map(flight=>{
var airline= flight.split(",")(4)
var delay = flight.split(",")(22)
(airline,delay)
})
以上獲得需要計算的相關字段。
val flightDelays=flightsMap.groupByKey()
val delayChance= flightDelays.map(airline=>{
var count=0
var totalCount =airline._2.size
for (delay<-airline._2){
if(delay.toInt>0){
count+=1
}
}
(airline._1,(count+0.0)/totalCount)
})
delayChance.take(20).foreach(println)
這里使用了scala的for循環，直接把value里面的list值輸入到一個變量delay里面去。