将列作为结构数组的 Spark 转储到镶木地板答案

【问题标题】：Spark dump to parquet with column as array of structures将列作为结构数组的 Spark 转储到镶木地板
【发布时间】：2020-03-03 10:29:02
【问题描述】：

我需要加载一个 csv 文件，该文件有一列具有结构数组，并将其以 parquet 格式转储到另一个位置。我的 csv 文件有两列，A 列和 B 列。 B列的数据类型为array<struct<x: bigint, y:bigint>>

我尝试使用如下架构加载 csv 文件：

val schemaB = ArrayType(StructType(Seq(StructField("x",LongType),StructField("y",LongType))))
val schema = new StructType().add("A",StringType).add("B",schemaB)
spark.read.option("sep", "\t").schema(schema).csv(<location>)

但是，这不起作用。我收到以下错误：

org.apache.spark.sql.AnalysisException: CSV data source does not support array<struct<x:bigint,y:bigint>&gt; data type.;</struct<x:bigint,y:bigint>

我什至尝试强制转换为所需的类型，但没有成功。

这是 B 列外观的示例：

|B                                                                                                                                                                                                                                                                                                                                                                                                                   |
+---------------------------------------------------------------------------------------------+
|68222:102332,21215:1000,10982:70330,|
|93302:13320,263721:902615,9382:100020,|

【问题讨论】：

您能分享您的示例输入数据吗？你怎么能在 csv 文件中有一个数组类型？
@koiralo 我已经添加了示例。 csv 会将其作为字符串。倾倒镶木地板时，我需要能够将其写为结构数组。
如果是这样，您可以将其读取为字符串，解析并转换为您想要的任何类型，然后将其写入 parquet。
是的，就像我在帖子中提到的那样，我尝试将其作为字符串读取并将其转换为数组，然后再写入镶木地板。那也没有用。我收到一个错误，无法将字符串转换为结构数组。

标签： scala apache-spark parquet

【解决方案1】：

如果您使用的是最新版本的 spark 即 2.4+

，您可以使用 transform 函数

首先以字符串形式读取，split 通过"," 获取列表，然后通过":" 再次读取split 以获取x 和y

val schema = new StructType().add("A",StringType).add("B",StringType)
val df = spark.read.option("delimiter", "\t").schema(schema).csv("path to csv")
val splitExpr =  expr("transform(split(B, ','), x -> (split(x, ':')[0] as x, split(x, ':')[1] as y))")

val result = df.select($"A", splitExpr.cast("array<struct<x: long, y:long>>") as "B" )

现在您可以将其保存在镶木地板中如果您使用的是旧版本的 spark，那么您需要编写一个 udf 最终架构：

root
 |-- A: string (nullable = true)
 |-- B: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- x: long (nullable = true)
 |    |    |-- y: long (nullable = true)

【讨论】：