为向量的数据帧生成编号规则时出错

Question

我使用rowid列为下表生成一个数字序列，以便执行联接，但这将引发以下错误。我做错了什么？请帮帮我。

fListVec: org.apache.spark.sql.DataFrame = [features: vector]
+-----------------------------------------------------------------------------+
|features                                                                     |
+-----------------------------------------------------------------------------+
|[2.5046410000000003,2.1487149999999997,1.0884870000000002,3.5877090000000003]|
|[0.9558040000000001,0.9843780000000002,0.545025,0.9979860000000002]          |
+-----------------------------------------------------------------------------+

代码：

import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row

val fListrdd = fListVec.rdd
    .map{case Row(features: Vector) => features}
    .zipWithIndex()
    .toDF("features","rowId")    

fListrdd.createOrReplaceTempView("featuresTable")
val f = spark.sql("SELECT features, rowId from featuresTable")
f.show(false)

输出：

import org.apache.spark.ml.linalg.vector import org.apache.spark.sql.Row org.apache.spark.sparkException：作业由于阶段失败而中止：阶段206.0中的任务0失败1次，最近的失败：阶段206.0中丢失的任务0.0（TID 1718，localhost,executor驱动程序）：Scala.MatchError：[[2.5046410000000003,2.148714999999997,1.0884870000000002,3.5877090000000003]]（属于类UNJOB$5。Apply(sparkcontext.scala:1944)在org.apache.spark.scheduler.resulttask.runtask.scala:87)在org.apache.spark.scheduler.task.run(task.scala:99)在org.apache.spark.executor.executor$taskrunner.run(executor.scala:282)在java.util.concurrent.threadpoolexecutor.runworker(threadpoolexecutor.1149)在scheduler.scala:1422)在org.apache.spark.scheduler.dagscheduler$$anonfun$handletasksetfailed$1。apply(dagscheduler.scala:802)在org.apache.spark.scheduler.dagscheduler$$anonfun$handletasksetfailed$1。apply(dagscheduler.scala:802)在scala.option.foreach（option.scala:257)在dd.zippedWithIndexRdd.(zippedWithIndexRdd.Scala:50)在org.apache.spark.rdd.rdd$$anonfun$zipwithIndex$1处。apply(rdd.scala:1293)在org.apache.spark.rdd.rdd$anonfun$zipwithIndex$1处。apply(Rdd.scala:1293)在org.apache.spark.rdd.rdd.rddoperationscope$.withscope(Rddoperationscope.scala:151)在ndexrdd.scala:52)在org.apache.spark.rdd.zippedWithindexrdd$$anonfun$2。apply(zippedWithindexrdd.scala:52)在org.apache.spark.sparkcontext$$anonfun$runjob$5。apply(sparkcontext.scala:1944)在org.apache.spark.sparkcontext$$anonfun$runjob$5。apply(sparkcontext.scala:1944)在

预期产出：

 features                 |       rowId

[2.5046410000000003,...]            0
[0.9558040000000001,...]            1

Answer 1

您就快到了--只需指定适当的向量类型densevector:

import org.apache.spark.sql.functions._
import org.apache.spark.mllib.linalg.DenseVector
import org.apache.spark.sql.Row

val fList = Seq(
  (Seq(2.5046410000000003, 2.1487149999999997, 1.0884870000000002, 3.5877090000000003)),
  (Seq(0.9558040000000001, 0.9843780000000002, 0.545025, 0.9979860000000002))
).toDF("features")

def seqToVec = udf(
  (s: Seq[Double]) => new DenseVector(s.toArray)
)

val fListVec = fList.withColumn("features", seqToVec($"features"))
// fListVec: org.apache.spark.sql.DataFrame = [features: vector]

val fListrdd = fListVec.rdd.
  map{ case Row(features: DenseVector) => features }.
  zipWithIndex.
  toDF("features", "rowId")  

fListrdd.show
// +--------------------+-----+
// |            features|rowId|
// +--------------------+-----+
// |[2.50464100000000...|    0|
// |[0.95580400000000...|    1|
// +--------------------+-----+

Answer 2

您必须在中间编写一个map函数，以便为要创建的新DataFrame定义数据类型

val fListrdd = fListVec.rdd
  .map{case Row(features) => features}
  .zipWithIndex()
  .map(x => (x._1.asInstanceOf[DenseVector], x._2.toInt))
  .toDF("features","rowId")

其中.map(x=>(x._1.asinstanceof[DenseVector],x.2.toint))行仅添加。

您可以进一步创建数据集。我个人推荐dataset，因为数据集是类型安全的，是数据流的优化形式。

为此，您需要一个案例类

case class features(features: DenseVector, rowId: Int)

并且只需在我的上述解决方案中添加featuresword，这样您就可以调用.todsapi来创建一个类型安全的DataSet。

val fListDS = fListVec.rdd
  .map{case Row(features: DenseVector) => features}
  .zipWithIndex()
  .map(x => features(x._1.asInstanceOf[DenseVector], x._2.toInt))
  .toDS