Мера искрового косинуса двух элементов

У меня есть файл .csv, который выглядит так:

message_id, hashtag
id_1, hashtag_1
id_2, hashtag_1
...
id_k, hashtag_m
....

Я пытаюсь найти меру косинуса между каждой парой хэштегов в CSV, выполнив следующие действия:

def isHeader(s: String):Boolean = {
    s.take(1) == "m"
}
def cs(pair: ( (String, String), (List[String], List[String]) ) ) = {
    val msgs1 = pair._2._1.toSet
    val msgs2 = pair._2._2.toSet
    val numer = msgs1.intersect(msgs2).size
    val denom = Math.sqrt(msgs1.size*msgs2.size)
    (pair._1._1, pair._1._2, numer / denom)
}
def to_csv_cos(t: (String, String, Double)): String = {
    t._1 ++ "," ++ t._2 ++ "," ++ t._3.toString
}
val messages = sc.textFile("....csv")
val msgData1 = messages.filter(x => !isHeader(x))
val data = msgData1.map(x => x.split(','))
val pairs = data.map(x => (x(0), List(x(1)))).reduceByKey((a,b) => a ++ b).flatMap(x => x._2.combinations(2).toList).map(x => (x(0), x(1)))
val msgs = data.map( x => (x(1), List(x(0)))).reduceByKey((a,b) => a++b)
val pairs_mapped = pairs.join(msgs).map{
     case (x, (y,z)) => (y, (x,z))
}.join(msgs).map{
    case (x, ( (y,z),t) ) => ( (x,y), (t,z) )
}
val res = pairs_mapped.map{
    x => cs(x)  
}.map(x => to_csv_cos(x)).saveAsTextFile("F:\\Scala\\result")

Идея такая:

1. Creating pairs from the elements (pairs)
2. Finding all messages for each hashtag (msgs)
3. Creating pairs: ( (hashtag_i, hashtag_j), (messages_with_hashtag_i, messages_with_hashtag_j) ) (pairs_mapped)
4. Calculating a measure(res)

Что ж, я думаю, мой код - просто мусор, поскольку я новичок в scala, spark и концепции функционального программирования, но он работает для небольшого csv (я пробовал со 100 строками). Но я должен рассчитать его для csv с ~ 25 миллионами строк, и в этом проблема. Процесс останавливается на saveAsTextFile (или SparkHadoopWriter.scala в пользовательском интерфейсе Spark) и не продолжается даже через 30 минут, а затем вылетает с различными ошибками (ошибки памяти, иногда просто «прерывание соединения» ...)

Я обнаружил на веб-сайте, что мы можем вычислить косинусную меру, используя фреймы данных, но я не совсем понимаю, как создать правильный фрейм данных из моих данных. Так что, пожалуйста, не могли бы вы дать мне несколько советов, как изменить мой код, чтобы он работал, или как создать правильный фрейм данных из CSV или чего-то еще?
Буду благодарен за любую помощь!