今天要从之前的数据中找出每个地点的一段时间内的产品发生率，即某地点有产量的天数/总天数。
在此之前，为了输出的可读性，我们需要用地点别名代替地点串号。
导入地点串号文件到hdfs中，串号文件每行格式如下：
1701170830479148,010102000000000000000013,1401010000000005,22,十区
其中第一列是串号，最后一列是地点别名
首先得到一个Map
val rfidCardFile = spark.sparkContext
.textFile("hdfs://Y40/medical_waste/card.csv")
.filter(line=>{
line.contains(","+ORG_ID+",")//过滤只有本机构内的记录
}).cache()
val rfidCardMap = rfidCardFile.map(line=>{
val a = line.split(",")
(a(0),a(4))
}).cache()
val localMap = rfidCardMap.collect().toMap
然后我们要构建一个机构内地点名称的数组
val teamWeightFile = spark.sparkContext.textFile("hdfs://Y40/medical_waste/team_mw.csv")
.cache()
val teamNameArray = teamWeightFile.map(line => {
val la = line.split(",")
localMap.get(la(0)).toString
}).distinct().collect()
然后我们需要构建一个地点的产出与否的数据源，这里可以简化成构建一个(地点,日期)数组，因为数据文件中记录的是产出日期和产出重量，没有产出就没有记录，重复的(地点，日期)只需要记录一次。
val teamDateArray = teamWeightFile.map(line => {
val la = line.split(",")
(localMap.get(la(0)).toString,la(1).substring(1,11).replaceAll("-","").toInt)
}).distinct().collect()
这里我们的日期使用整型记录，类似于20170227这样。
接着我们需要知道两个日期之间的天数，分为两个函数，首先把上述整型表述的时间转为LocalDateTime，然后再使用Duration.between计算。
def intToLocalDate(day1:Int):LocalDateTime = {
val day1D = day1 %100
val day1M = ((day1 - day1D)/100) %100
val day1Y = (day1 - day1D - (day1M*100))/10000
LocalDateTime.of(day1Y,day1M,day1D,0,0,0)
}
def countDay(day1:Int,day2:Int):Long = {
val day1Date =intToLocalDate(day1)
val day2Date =intToLocalDate(day2)
Duration.between(day2Date,day1Date).abs().toDays
}
最后遍历teamNameArray
teamNameArray.map(teamName=>{
val happenCount = teamDateArray.filter(item=>{
item._1==teamName && item._2>=BEGIN_DATE && item._2<END_DATE
}).length
val result = teamName+"\t"+happenCount+"\t"+dayLength+"\t"+(happenCount.toDouble/dayLength.toDouble)
println(result)
})
这里BEGIN_DATE和END_DATE可以自己定义，只要是整型表述的日期即可。
为什么是本地的map和array呢？如果使用rdd的话，性能是否会更好？答案很遗憾，如果使用rdd来做rfidCardMap和teamNameArray，编译虽然会通过，运行却会报错，大意是rdd并不支持嵌套操作，也就是说不能在一个rdd的操作因子中使用到另外一个rdd。
