關於spark 矩陣轉置的一個坑,解決了
下面是我用scala寫的一個spark程式。輸入檔案是一個類似下面的表格檔案使用tab隔開 想把行變成列 順序不變第一行就是第一列依次類推
XZ274 BJXHBAO BJXHWU BJWJ10 BJWJ16 BJWJ17 wls-5-18
106 C C C C T C C C C C C T C C
108 A A A A G A A C C C C T C C
111 C C T C C C T C C C C T C C
125 T T C T T C C C C C C T C C
126 G G G G G G G C C C C T C C
128 G G G G G G G C C C C T C C
129 C C C C C C C C C C C T C C
132 T T T T T T T C C C C T C C
133 T T T T T T T C C C C T C C
135 A G G A G G G C C C C T C C
138 C C C C C C C C C C C T C C
我想把行變列用矩陣的轉置實現了。所以使用了CoordinateMatrix來處理 AGTC 用數字代表了,最後在轉化回來就可以了,但是我發現,資料在一定的行數的時候必須分割槽足夠大,否則行與行就會序列 也就是說本來在第四行的向量跑到了第一行 。但是行裡面的元素沒有錯,就是位置錯了,只要將分割槽增大,就會沒有問題了。到現在還是沒有想明白是怎麼回事,分割槽要是1 直接就完全錯亂了
下面的程式碼思路也很簡單,先把第一行過濾了,然後根據行標建立一個 CoordinateMatrix 矩陣(行列在這個時候就自動變換了,其實就是原來的行列號相互交換以一下就可以了),這個時候由於map的時候CoordinateMatrix的 i,j 已經指定好了,所以直接轉化成 toIndexedRowMatrix 其實RowMatrix 也可以,然後就可以直接輸出了,但是分割槽設小了 ,就會錯亂,設定大點就會沒有問題
後記:經過研究發現原來 RowMatrix 行號資訊丟失了,所以後來完全就是隨機了,我去!!!!!大概是這樣,要是有問題請大家指教
只要修改下程式碼就可以了程式碼如下:
var coord=new CoordinateMatrix(flatmap);
var indexmx=coord.toIndexedRowMatrix();
var tmp=indexmx.rows.map(item=>{
var index=item.index
Tuple2(index,item.vector);
})
var result=tmp.sortByKey();
是的。。。就是按照行號排序,反正我是可以了。要是直接map IndexedRowMatrix 裡面的item還是不按照順序來的,實在是不明白怎麼會是,感覺還有別的坑。。。。。
object TMain {
def main(args:Array[String])={
var conf=new SparkConf();
var spark=new SparkContext(conf);
var raw_data=spark.textFile(args(0),args(2).toInt).zipWithIndex();
var newtext=raw_data.filter{
case(line,index)=>{
var token=line.split('\t');
if(token(0).isEmpty())
false;
else
true;
}};
//A=1 G=2 T=3 C=4
var count=newtext.map{case(item,index)=>{
var token=item.split('\t');
var i=0;
var colmn=0;
var set:List[MatrixEntry]=List();
for(key <-token)
{
if(i==0)
{
i=1; //跳過第一列
}
else
{
key match {
case "A"=>{set=set.+:(MatrixEntry.apply(colmn.toLong, index-1, 1.0));}
case "G"=>{set=set.+:(MatrixEntry.apply(colmn.toLong, index-1, 2.0));}
case "T"=>{set=set.+:(MatrixEntry.apply(colmn.toLong, index-1, 3.0));}
case "C"=>{set=set.+:(MatrixEntry.apply(colmn.toLong, index-1, 4.0));}
}
colmn=colmn+1;
}
}
set
}
};
var flatmap=count.flatMap(item=>{
item
})
flatmap.cache();
var coord=new CoordinateMatrix(flatmap);
var mx=coord.toIndexedRowMatrix();
var str=mx.rows.map(x=>{
var str=new StringBuilder();
var ve=x.vector;
for(c <- ve.toArray)
{
c match {
case 0.0=>{}
case 1.0=>{str.append("A");}
case 2.0=>{str.append("G");}
case 3.0=>{str.append("T");}
case 4.0=>{str.append("C");}
}
}
str.toString()
});
str.saveAsTextFile(args(1));
}
}