Spark - 将矢量数据拆分、转换和存储为 CSV 文件答案

【问题标题】：Spark - Splitting, Converting and Storing Vector Data into a CSV fileSpark - 将矢量数据拆分、转换和存储为 CSV 文件
【发布时间】：2018-02-15 06:23:55
【问题描述】：

我有一个 Spark 数据框，其中一个列中包含矢量数据，如下所示的第一列

+--------------------+-----+-----------+
|            features|Label|OutputLabel|
+--------------------+-----+-----------+
|(1133,[33,296,107...|    0|        0.0|
|(1133,[19,1045,10...|    0|        0.0|
|(1133,[9,398,1075...|    0|        0.0|
|(1133,[0,927,1074...|    0|        0.0|
|(1133,[41,223,107...|    0|        0.0|
|(1133,[70,285,108...|    0|        0.0|
|(1133,[4,212,1074...|    0|        0.0|
|(1133,[25,261,107...|    0|        0.0|
|(1133,[0,258,1074...|    0|        0.0|
|(1133,[2,219,1074...|    0|        0.0|
|(1133,[8,720,1074...|    0|        0.0|
|(1133,[2,260,1074...|    0|        0.0|
|(1133,[54,348,107...|    0|        0.0|
|(1133,[167,859,10...|    0|        0.0|
|(1133,[1,291,1074...|    0|        0.0|
|(1133,[1,211,1074...|    0|        0.0|
|(1133,[23,216,107...|    0|        0.0|
|(1133,[126,209,11...|    0|        0.0|
|(1133,[70,285,108...|    0|        0.0|
|(1133,[96,417,107...|    0|        0.0|
+--------------------+-----+-----------+

请看下面这个数据框的架构

root
 |-- features: vector (nullable = true)
 |-- Label: integer (nullable = true)
 |-- OutputLabel: double (nullable = true)

问题 1：我需要将第一列数据拆分为两列，以便整数数据应位于一列，数组数据应位于另一列。不知道如何在 Spark / Scala 中做到这一点？对此的任何指示都会有所帮助。

当我尝试将此数据帧写入 csv 文件时，出现以下错误

线程“主”java.lang.UnsupportedOperationException 中的异常： CSV 数据源不支持结构，值：数组> 数据类型。
问题 2：我知道即使这个数据帧也不能作为文本文件写入，因为它只会将一列写入输出文件并且它不应该是 Struct 类型。那么是否可以在将第一列分成两个单独的列后写入这个数据框？第二列数据将是数组数据类型。我们可以这样写入输出文件吗？
问题 3：我们是否可以将数组数据单独写入 csv 文件？

【问题讨论】：

你能分享你的数据框的架构吗？
@RameshMaharjan 我已将问题更新为具有架构。请检查一下

标签： apache-spark apache-spark-sql

【解决方案1】：

那么是否可以在将第一列分成两个单独的列后写入此数据帧？

没有。你看到的只是SparseVector 的代表。即使您提取索引和值，CSV source 也仅支持原子类型。

如果您不喜欢使用 CSV，我会将一整列转换为 JSON

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.functions._

val df = sc.parallelize(Seq(
  (Vectors.sparse(100, Array(1, 11, 42), Array(1, 2, 3)), 0, 0.0)
)).toDF("features", "label", "outputlabel")

df.withColumn("features", to_json(struct($"features"))).write.csv(...)

要将其解析为Vector，请按照说明provided here

【讨论】：