记一次spark数据倾斜的优化

val featureTime = timeList
.map { t1 => (Math.abs(t1 - sampleTime), t1) }
.minBy(v => v._1)._2

val featureTime = biSearch(featureTimeList, sampleTime)
def biSearch(data:Array[Long], input:Long):LOng= {
var start = 0
var end = data.length - 1
var retIndex = 0
if(data(start) <= input ){
data(start)
}else if(data(end) >= input){
data(end)
}else{
while(start <end){
val index = (start + end)/2
if(data(index) > input && data(index+1) > input){
start = index
}else if(data(index) data(index + 1)
end = index
}else if(data(index) > input && data(index + 1)
val c = data(index) + data(index + 1) -2*input
if(c > 0){
retIndex = index + 1
}else{
retIndex = index
}
start = end
}else if(data(index) == input){
retIndex = index
start = end
}else if(data(index + 1) == input){
retIndex = index + 1
start = end
}
}
data(retIndex)
}
}

def searchTimeKey_user(keyArrayJson: RDD[(String, (Array[String], JSONObject))],
onlineTime: RDD[(String, Array[Long])]): RDD[(String, (Array[String], JSONObject))] = {
keyArrayJson.leftOuterJoin(onlineTime, 100)
.map { case (id, ((arr, itemFeature), timeList)) =>
val t0 = TimeUtil.getSecondTimeStamp(arr(3))
var key = String.valueOf(Random.nextInt(100)) // 当关联不上的时候key反正没用，打散了防止数据倾斜
if (timeList.isDefined) {
if(timeList.get.length > 10){
val t1 = biSearch(timeList.get, t0)
key = id + String.valueOf(t1)
}else{
val t1 = timeList.get.map { t1 => (Math.abs(t1 - t0), t1) }.minBy(v => v._1)._2
key = id + String.valueOf(t1)
}
}
(key, (arr, itemFeature))
}
}