0

使用 Spark 2.1.1，我有一个 N 行 csv 作为 'fileInput'

colname datatype    elems   start   end
colA    float       10      0       1
colB    int         10      0       9

我已经成功地制作了一个 sql.rows 数组...

val df = spark.read.format("com.databricks.spark.csv").option("header", "true").load(fileInput)
val rowCnt:Int = df.count.toInt
val aryToUse  = df.take(rowCnt)
Array[org.apache.spark.sql.Row] = Array([colA,float,10,0,1], [colB,int,10,0,9])

针对这些行并使用我的随机值生成器脚本，我成功填充了一个空的 ListBuffer[Any] ...

res170: scala.collection.mutable.ListBuffer[Any] = ListBuffer(List(0.24455154, 0.108798146, 0.111522496, 0.44311434, 0.13506883, 0.0655781, 0.8273762, 0.49718297, 0.5322746, 0.8416396), List(1, 9, 3, 4, 2, 3, 8, 7, 4, 6))

现在，我有一个具有不同类型列表的混合类型 ListBuffer[Any]。. 如何迭代和压缩这些？ [任何] 似乎无视映射/压缩。我需要获取由 inputFile 的定义生成的 N 个列表，然后将它们保存到 csv 文件中。最终输出应该是：

ColA, ColB
0.24455154, 1
0.108798146, 9
0.111522496, 3
... etc

然后 inputFile 可用于创建任意数量的“colnames”，任何“数据类型”（我有脚本），每种类型出现 1::n 次，任意数量的行（定义为“elems”）。我的随机生成脚本自定义每个“开始”和“结束”的值，但这些列与这个问题无关）。

4

2 回答 2

2

给定 a List[List[Any]]，您可以使用 将所有这些列表“压缩”在一起transpose，如果您不介意结果是列表列表而不是元组列表：

val result: Seq[List[Any]] = list.transpose

如果您想将其写入 CSV，您可以首先将每个“行”映射为逗号分隔的字符串：

val rows: Seq[String] = result.map(_.mkString(","))

（注意：我忽略了 Apache Spark 部分，这似乎与这个问题完全无关......“元数据”是通过 Spark 加载的，但随后它被收集到一个数组中，所以它变得无关紧要）

于 2017-05-18T18:45:58.317 回答
-1

我认为RDD.zipWithUniqueId()orRDD.zipWithIndex()方法可以执行您想做的事情。

更多信息请参考官方文档。希望这对你有帮助

于 2017-05-18T18:20:10.350 回答