Доступ к кортежу внутри кортежа для задания анонимной карты в Spark

Question

Этот пост в основном о том, как строить суставные и маргинальные гистограммы из (String, String) RDD. Я отправил код, который я в конечном итоге использовал ниже в качестве ответа.

у меня есть RDD который содержит набор кортежей типа (String,String) и поскольку они не уникальны, я хочу посмотреть, сколько раз встречается каждая строка, поэтому я использую countByValue вот так

val PairCount = Pairs.countByValue().toSeq

который выдает мне кортеж в качестве вывода (this ((String,String),Long), где long - количество раз, когда появился кортеж (String, String)

Эти строки могут повторяться в разных комбинациях, и я, по сути, хочу запустить подсчет слов в этой переменной PairCount, поэтому я попробовал что-то вроде этого, чтобы начать:

PairCount.map(x => (x._1._1, x._2))

Но вывод, который это выдает, это String1->1, String2->1, String3->1 и т. Д.

Как вывести пару ключ-значение из задания карты в этом случае, когда ключом будет одно из значений String из внутреннего кортежа, а значением будет значение Long из внешнего кортежа?

Обновление: @vitalii получает меня почти там. ответ заставляет меня перейти к Seq[(String,Long)], но мне действительно нужно превратить это в карту, чтобы я мог потом запустить ее при помощи limitByKey. когда я бегу

PairCount.flatMap{case((x,y),n) => Seq[x->n]}.toMap

для каждого уникального х я получаю х-> 1

например, приведенная выше строка кода генерирует mom->1 dad->1, даже если включены кортежи из flatMap (mom,30) (dad,59) (mom,2) (dad,14), и в этом случае я бы ожидайте, что Карта предоставит маму->30, папа->59, мама->2 папа->14. Тем не менее, я новичок в Scala, поэтому я могу неправильно истолковывать функциональность.

как я могу преобразовать последовательность Tuple2 в карту, чтобы уменьшить ее на ключах карты?

Answer 1

Чтобы получить гистограммы для (String,String) RDD, я использовал этот код.

val Hist_X  = histogram.map(x => (x._1-> 1.0)).reduceByKey(_+_).collect().toMap
val Hist_Y  = histogram.map(x => (x._2-> 1.0)).reduceByKey(_+_).collect().toMap
val Hist_XY = histogram.map(x => (x-> 1.0)).reduceByKey(_+_)

где гистограмма была (строка, строка), СДР

Answer 2

Если я правильно понимаю вопрос, вам нужен flatMap:

val pairCountRDD = pairs.countByValue() // RDD[((String, String), Int)]
val res : RDD[(String, Int)] = pairCountRDD.flatMap { case ((s1, s2), n) =>
   Seq(s1 -> n, s2 -> n)
}

Обновление: я не совсем понял, какова ваша конечная цель, но вот еще несколько примеров, которые могут вам помочь, кстати, приведенный выше код неверен, я упустил тот факт, что countByValue возвращает карту, а не RDD:

val pairs = sc.parallelize(
  List(
    "mom"-> "dad", "dad" -> "granny", "foo" -> "bar", "foo" -> "baz", "foo" -> "foo"
  )
)
// don't use countByValue, if pairs is large you will run out of memmory
val pairCountRDD = pairs.map(x => (x, 1)).reduceByKey(_ + _) 

val wordCount = pairs.flatMap { case (a,b) => Seq(a -> 1, b ->1)}.reduceByKey(_ + _)

wordCount.take(10)

// count in how many pairs each word occur, keys and values:
val wordPairCount = pairs.flatMap { case (a,b) => 
               if (a == b) {
                 Seq(a->1)
               } else {
                  Seq(a -> 1, b ->1)
               }
             }.reduceByKey(_ + _)
wordPairCount.take(10)