Итерация над элементом RDD

Question

У меня есть RDD этого типа: (String, (Iterable [(String, Int)], Double)). Я хотел бы преобразовать это, чтобы иметь новый RDD следующего типа: (String, (Iterable [(String, Int)], Double)), где каждый Int в исходном итерабельном умножается на Double. Как это можно сделать? Большинство моих аттапттов привели к тому, что УРД не имеют реальных данных.

Answer 1

Это похоже на то, что вы хотите. Он компилируется, у меня нет набора данных для его запуска.

def method(src: RDD[(String, (Iterable[(String, Int)], Double))]) = { 
    src.map { 
     case (s, (iter, d)) => 
     (s, (iter.map { case (s, i) => (s, (i * d).toInt) }, d)) 
    } 
    } 

Или mapValues ​​может быть более эффективной, так как он не влияет на разбиение:

def method(src: RDD[(String, (Iterable[(String, Int)], Double))]) = { 
    src.mapValues { 
     case (iter, d) => 
     (iter.map { case (s, i) => (s, (i * d).toInt) }, d) 
    } 
    } 

Answer 2

Вы можете RDD.mapValues, которые будут применяться данной функцией для всех значений с ключами постоянны.

def transformRdd(in: RDD[(String, (Iterable[(String, Int)], Double))]) := RDD[(String, (Iterable[(String, Int)], Double))]{ 
    in.mapValues{ 
//Here a will be iterable and b is Double 
    case (a, b) => (a.map { case (str, int) => (str, (int * b).toInt) }, b) 
    } 

Теперь это сработает ... Проверьте, не скомпилировано ли это.