我正在嘗試轉置一個巨大的資料幀（100Mx20K）。由于資料幀分布在多個節點上并且難以在驅動程式上收集，我想通過mllib矩陣轉換來進行轉置。這個想法似乎已經在別處進行了測驗，所以選擇的程序如下：
import org.apache.spark.sql.functions._
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.{IndexedRow, IndexedRowMatrix}
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.parquet("temp/test.parquet").select("H1","H2","H3","H4")
val matrixColumns = df.columns
val rdd = df.select(array(matrixColumns:_*).as("arr")).as[Array[Int]].rdd
.zipWithIndex()
.map{ case(arr, index) => IndexedRow(index, Vectors.dense(arr.map(_.toDouble)))}
val dm = new IndexedRowMatrix(rdd).toBlockMatrix().toLocalMatrix()
我注意到一種可能的型別并嘗試替換：
orig:
val rdd = df.select(array(matrixColumns:_*).as("arr"))....
modified:
val rdd = df.select(Array(matrixColumns:_*)).as("arr")...
但是，兩者都不適合我，并且上述更改會引發錯誤：
scala> df.select(Array(matrixColumns:_*)).as("arr")
^
error: overloaded method select with alternatives:
[U1](c1: org.apache.spark.sql.TypedColumn[org.apache.spark.sql.Row,U1]): org.apache.spark.sql.Dataset[U1] <and>
(col: String,cols: String*)org.apache.spark.sql.DataFrame <and>
(cols: org.apache.spark.sql.Column*)org.apache.spark.sql.DataFrame
cannot be applied to (Array[String])
我不確定是否存在版本問題（我使用的是 Spark 3.3.0）或者問題是否出在其他地方。對于修復上述錯誤的任何幫助，我將不勝感激。
uj5u.com熱心網友回復：
將選擇呼叫更改為：
df.select(matrixColumns.head, matrixColumns: _*)
或者
import org.apache.spark.sql.functions.col
df.select(matrixColumns.map(col(_)):_*)
轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/508584.html