spark scala Apriori

val mydata = Array(Array(1,3,4,5),Array(2,3,5),Array(1,2,3,4,5),Array(2,3,4,5))
val pamydata = sc.parallelize(mydata)
val C1 = pamydata.flatMap(_.toSet).distinct().collect().map(Set(_))
val D = mydata.map(_.toSet)
val D_bc = sc.broadcast(D)
val length = mydata.length
var limit = 0.70
def f1(a:Set[Int],B:Array[Set[Int]],length:Int,limit:Double) = {if(B.filter(b => a.subsetOf(b)).size/length.toDouble >= limit)(a,B.filter(b => a.subsetOf(b)).size/length.toDouble)}
var suppdata = sc.parallelize(C1).map(f1(_,D_bc.value,4,limit)).filter(_.!=(())).collect()
var L = Array[Array[Set[Int]]]()
val L1 = suppdata.map(_ match{case a:Tuple2[_,_] => a._1 match{ case b:Set[_] => b.asInstanceOf[Set[Int]]}})
L = L :+ L1
var k=2
while(L(k-2).length>0){
var CK = Array[Set[Int]]()
for((var1,index) <- L(k-2).zipWithIndex;var2 <- L(k-2).drop(index+1) if var1.take(k-2).equals(var2.take(k-2))){CK= CK :+ (var1|var2)}
val suppdata_temp = sc.parallelize(CK).map(f1(_,D_bc.value,4,limit)).filter(_.!=(())).collect()
suppdata = suppdata :+ suppdata_temp
L = L :+ suppdata_temp.map(_ match{case a:Tuple2[_,_] => a._1 match{ case b:Set[_] => b.asInstanceOf[Set[Int]]}})
k += 1
}
L = L.filter(_.nonEmpty)
L
suppdata