Интервал слияния с использованием Spark Scala - PullRequest
Новая
       15

Интервал слияния с использованием Spark Scala

2 голосов
/ 29 мая 2019

У меня есть список интервалов, которые я хотел бы объединить при наложении.

пример: List((1,1),(2,2),(4,4),(5,5)) Требуемый вывод здесь List((1,2),(4,5))

У меня есть список чисел на 2,5 ГБ, который я хотел бы преобразовать в диапазоны.

Примечание: в списке ввода нет дубликатов

Шаги

  1. input: List [Int].
  2. сопоставить со списком кортежей: список ((a, a), (b, b), ...).
  3. уменьшить его с помощью логики слияния диапазонов.
val l = List(List((1,1)),List((2,2)),List((4,4)),List((5,5)))
val list =sc.parallelize(l)

def merge(r1:(Int,Int),r2:(Int,Int))  :(Int,Int)  = {
    if(r1._2+1==r2._1) (r1._1,r2._2)
    else if(r2._2+1 == r1._1) (r2._1,r1._2)
    else null
}

val res = list.reduce((x,y) =>{
   x.map(r1 => {
        y.map(r2 => {
            val m = merge(r1,r2)
             m match {
                case null => List(r1,r2)
                case _ => List(m)
             }
         }).flatten
    }).flatten
})

res: List[(Int, Int)] = List((4,5), (2,2), (1,2))

Фактический результат равен res: List[(Int, Int)] = List((4,5), (2,2), (1,2)), где, как я ожидаю, List((4,5),(1,2)).

edit: мое решение

Я попробовал следующий код.Кажется, работает с небольшим вводом, но занимает слишком много времени для моих исходных данных.Есть ли лучшее решение, чем это? 

def overlap(x: (Int,Int),y:(Int,Int)) = {
    if(x._2==y._1) (x._1,y._2)
    else if(x._1==y._2) (y._1,x._2)
    else null
}

def isOverlapping(x: (Int,Int),y:(Int,Int)) = {
    x._1 == y._1 || x._1 == y._2 || x._2==y._1 || x._2==y._2 
}

val res = list.reduce((x,y) =>{
  val z =  x.map(r1 => {
        y.map(r2 => {
            val m = merge(r1,r2)
             m match {
                case null => List(r1,r2)
                case _ =>{
                     List(m)
                }
             }
         }).flatten
    }).flatten

//-------compressing the accumulated list z to merge overlapping tuples

    z.foldLeft(List[(Int,Int)]()) { (acc, i) => {
    if (!acc.exists(isOverlapping(i, _)))
        i +: acc
      else
        acc.map(x => {
            val m = overlap(x,i)
             m match {
                case null => x
                case _ => m
             }
        })
    }}
//---------

})


res: List[(Int, Int)] = List((4,5), (1,2))
...