【问题标题】:How to flatten RDD of sequences如何展平序列的 RDD
【发布时间】:2019-11-28 19:37:45
【问题描述】:

我目前有一个RDD[Seq[MatrixEntry]],我正试图通过展开或展平Seq 将其转换为RDD[MatrixEntry]。我可以将Seq 转换为其他数据类型,例如List,但我无法完全删除它。

我的问题与this 惊人地相似,因为我可以转换Seq,但使用MatrixEntry 而不是Ints,但尝试类似的解决方案会产生

error: type mismatch;
 found   : org.apache.spark.mllib.linalg.distributed.MatrixEntry
 required: scala.collection.GenTraversableOnce[?]

我尝试运行的代码是尝试将 DataFrame 转换为 CoordinateMatrix:

def matmaker(data: DataFrame):Unit={
    val rrd:RDD[(List[(Any,Int)],Long)] = data.rdd
        .map(r => r.toSeq.toList.zipWithIndex)
        .zipWithIndex()
    val precord:RDD[MatrixEntry] = rrd.map{
      case ((s:Seq[(Any,Int)],r:Long)) => s.map{
        case (value:Any,c:Int) => MatrixEntry(r,c,value.toString.toDouble)
      }.flatMap(List => List) //Problem Here
    }
    precord.foreach(println)
    //new CoordinateMatrix(precord.map(r=>r.take(r.length-1)))
  }

完成测试后,函数的最终类型将从Unit 更改为CoordinateMatrix

【问题讨论】:

  • 您能否向我们展示您正在运行的带有类型注释的代码?
  • @Ethan 刚刚编辑
  • 更改此:val precord:RDD[MatrixEntry] = rrd.map{ 为:val precord:RDD[MatrixEntry] = rrd.flatMap { 并删除此:.flatMap(List => List)
  • @LuisMiguelMejíaSuárez 成功了!太感谢了。您想将其发布为答案,以便我可以信任您吗?
  • @Cheezbeez neh,不用担心,这只是一个新手错误。永远记住,如果你想展平函数的结果使用flatMap,否则使用map;)

标签: scala apache-spark sequence rdd


【解决方案1】:

您的平面图只是放错了位置。

你需要写}}.flatMap(x=>x)而不是}.flatMap(x=>x)},像这样:

def matmaker(data: DataFrame):CoordinateMatrix={
    val rrd:RDD[(List[(Any,Int)],Long)] = data.rdd
        .map(r => r.toSeq.toList.zipWithIndex)
        .zipWithIndex()
    val precord:RDD[MatrixEntry] = rrd.map{
        case ((s:Seq[(Any,Int)],r:Long)) => s.map{
            case (value:Any,c:Int) => MatrixEntry(r,c,value.toString.toDouble)
        }//No Problem Here, flatmap was moved
    }.flatMap(x => x) 
    new CoordinateMatrix(precord)
}

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 2014-12-17
    • 1970-01-01
    • 1970-01-01
    • 2019-11-20
    • 2019-08-08
    • 1970-01-01
    • 2016-06-24
    • 1970-01-01
    相关资源
    最近更新 更多