從RDD

Question

創建2D矩陣我有類型的以下RDD（（用戶ID，MovieID），1）：

val data_wo_header=dropheader(data).map(_.split(",")).map(x=>((x(0).toInt,x(1).toInt),1)) 

欲該數據結構轉換成一個二維數組，使得所有元素（用戶ID電影ID）有一個1 else 0

我認爲我們必須映射用戶ID爲0-N，如果N是不同用戶的數量並將電影ID映射到0-M如果不是不同電影的數量。

編輯：如你提到的例子

 Movie ID-> 

Userid 1 2 3 4 5 6 7 

1  0 1 1 0 0 1 0 
2  0 1 0 1 0 0 0 
3  0 1 1 0 0 0 1 
4  1 1 0 0 1 0 0 
5  0 1 1 0 0 0 1 
6  1 1 1 1 1 0 0 
7  0 1 1 0 0 0 0 
8  0 1 1 1 0 0 1 
9  0 1 1 0 0 1 0 

The RDD will be of the sort 
(userID, movID,rating) 
101,1002,3.5 
101,1003,2.5 
101,1006,3 
102,1002,3.5 
102,1004,4.0 
103,1002,1.0 
103,1003,1.0 
103,1007,5.0 
…. 

Answer 1

HI我管理使用下面的函數來生成的2D矩陣。這需要在格式

((userID, movID),rating) 
101,1002,3.5 
101,1003,2.5 
101,1006,3 
102,1002,3.5 
102,1004,4.0 
103,1002,1.0 
103,1003,1.0 
103,1007,5.0 

的RDD並返回特徵矩陣：

def generate_characteristic_matrix(data_wo_header:RDD[((Int, Int), Int)]):Array[Array[Int]]={ 
    val distinct_user_IDs=data_wo_header.sortByKey().map(x=>x._1._1).distinct().collect().sorted 
    val distinct_movie_IDs=data_wo_header.sortByKey().map(x=>x._1._2).distinct().collect().sorted 

    var movie_count=distinct_movie_IDs.size 
    var user_count=distinct_user_IDs.size 

    var a =0 
    var map_movie = new ArrayBuffer[(Int, Int)]() 
    var map_user = new ArrayBuffer[(Int, Int)]() 
    //map movie ID's from (0,movie_count) 
    for(a <- 0 to movie_count-1){ 
     map_movie+=((distinct_movie_IDs(a),a)) 
    } 
    //map user ID's from (0,user_count) 
    for(a <- 0 to user_count-1){ 
     map_user+=((distinct_user_IDs(a),a)) 
    } 
    //size of char matrix is user_countxmovie_count 
    var char_matrix = Array.ofDim[Int](user_count,movie_count) 
    data_wo_header.collect().foreach(x => { 
     var user =x._1._1 
     var movie=x._1._2 
     var movie_mappedid=map_movie.filter(x=>x._1==movie).map(x=>x._2).toArray 
     var user_mappedid=map_user.filter(x=>x._1==user).map(x=>x._2).toArray 
     char_matrix(user_mappedid(0))(movie_mappedid(0))=1 
    }) 
    return char_matrix 
    } 

Answer 2

val baseRDD = sc.parallelize(Seq((101, 1002, 3.5), (101, 1003, 2.5), (101, 1006, 3), (102, 1002, 3.5), (102, 1004, 4.0), (103, 1002, 1.0), (103, 1003, 1.0), (103, 1007, 5.0)))  
     baseRDD.map(x => (x._1, x._2)).groupByKey().foreach(println) 

（用戶ID，movID，等級）格式

結果：

（101，CompactBuffer（1002，1003，1006 ））

（102，CompactBuffer（1002,1004））

（103，CompactBuffer（1002，1003，1007））