Spark 2.0.x从包含一个string类型数组的数据框中转储csv文件

Question

我有一个数据框df,其中包含一列类型数组

df.show() 好像

|ID|ArrayOfString|Age|Gender|
+--+-------------+---+------+
|1 | [A,B,D]     |22 | F    |
|2 | [A,Y]       |42 | M    |
|3 | [X]         |60 | F    |
+--+-------------+---+------+

我尝试将其转储到dfcsv文件中,如下所示:

val dumpCSV = df.write.csv(path="/home/me/saveDF")

由于该列,它无法正常工作ArrayOfString.我收到错误:

CSV数据源不支持数组字符串数据类型

如果我删除列,代码将起作用ArrayOfString.但我需要保持ArrayOfString!

转储csv数据帧的最佳方法是什么,包括列ArrayOfString(ArrayOfString应该作为一个列转储到CSV文件中)

Answer 1

您收到此错误的原因是csv文件格式不支持数组类型,您需要将其表示为能够保存的字符串.

请尝试以下方法:

import org.apache.spark.sql.functions._

val stringify = udf((vs: Seq[String]) => vs match {
  case null => null
  case _    => s"""[${vs.mkString(",")}]"""
})

df.withColumn("ArrayOfString", stringify($"ArrayOfString")).write.csv(...)

要么

import org.apache.spark.sql.Column

def stringify(c: Column) = concat(lit("["), concat_ws(",", c), lit("]"))

df.withColumn("ArrayOfString", stringify($"ArrayOfString")).write.csv(...)

Answer 2

Pyspark实施:

在这个例子中,字段更改column_as_array到column_as_string保存之前.

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def array_to_string(my_list):
    return '[' + ','.join([str(elem) for elem in my_list]) + ']'

array_to_string_udf = udf(array_to_string,StringType())

df = df.withColumn('column_as_str',array_to_string_udf(d["column_as_array"]))

然后您可以在保存之前删除旧列(数组类型).

df.drop("column_as_array").write.csv(...)

Answer 3

如果您已经知道哪些字段包含数组，则不需要UDF。您可以简单地使用Spark的cast函数：

    val dumpCSV = df.withColumn("ArrayOfString", lit("ArrayOfString).cast("string"))
                    .write
                    .csv(path="/home/me/saveDF"

)

希望能有所帮助。

Answer 4

这是将a的所有ArrayType（任何基础类型）列转换DataFrame为StringType列的方法：

def stringifyArrays(dataFrame: DataFrame): DataFrame = {
  val colsToStringify = dataFrame.schema.filter(p => p.dataType.typeName == "array").map(p => p.name)

  colsToStringify.foldLeft(dataFrame)((df, c) => {
    df.withColumn(c, concat(lit("["), concat_ws(", ", col(c).cast("array<string>")), lit("]")))
  })
}

另外，它不使用UDF。