如何更改数组中的 Spark Dataframe 列数据类型

Question

关于我的一个更大的问题，我遇到了两个小问题：我想每天读取一次 JSON 数据并将其保存为 Parquet 以供以后与数据相关的工作。使用镶木地板要快得多。但是我遇到的问题是，在读取该 parquet 时，Spark 总是尝试从模式文件中获取模式，或者只是从第一个 parquet 文件中获取模式，并假定所有文件的模式都是相同的。但有些情况下，我们在某些列中几天没有任何数据。

假设我有一个 JSON 文件，其中的数据具有以下架构：

root
 |-- Id: long (nullable = true)    
 |-- People: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- Name: string (nullable = true)
 |    |    |-- Amount: double (nullable = true)

然后我有另一个 JSON 文件，其中没有“人员”列的数据。因此架构如下：

root
 |-- Id: long (nullable = true)    
 |-- People: array (nullable = true)
 |    |-- element: string (containsNull = true)

当我将它们与read.json 一起读入时，Spark 会遍历所有文件并从中推断出合并的架构，更具体地说是从第一个文件中推断出合并的架构，然后将第二个文件中的行留空，但架构是正确。

但是当我分别阅读这些内容并分别写入 parquet 时，我无法将它们一起阅读，因为对于 Parquet，架构不匹配并且我收到错误消息。

我的第一个想法是读取缺少数据的文件并通过强制转换列类型以匹配第一个模式来手动更改其模式，但是这种手动转换是错误的，它可能不同步，我什至不知道如何将此字符串类型转换为数组或结构类型。

另一个问题是当“金额”字段只有完整的整数时，Spark 会根据需要将它们读取为 longs 而不是 doubles。但如果我使用：

val df2 = df.withColumn("People.Amount", col("People.Amount").cast(org.apache.spark.sql.types.ArrayType(org.apache.spark.sql.types.DoubleType,true)))

那么它并没有改变原来列的类型，而是增加了一个名为People.Amount的新列

Answer 1

我认为您可以通过模式合并来解决问题（请参阅文档here）。如果您拥有的第一个镶木地板具有正确的架构，那么您可以执行类似的操作将该架构应用于新的实木复合地板吗？

// Read the partitioned table
val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()

编辑

你说有 200 多列，你都知道了吗？我看到了两条前进的道路，并且可能有很多方法可以实现这一目标。一种是您预先定义所有可以看到的字段。我过去所做的是创建一个带有单个虚拟记录的 json 文件，其中包含我想要的所有字段，并且完全按照我想要的方式输入。然后，您可以始终与“星期一”或“星期二”数据集同时加载该记录，并在加载后将其删除。这可能不是最佳实践，但这就是我在前进的道路上跌跌撞撞的方式。

另一种方法是停止尝试在正确的架构中加载/保存单个数据集，并在加载所有数据后设置架构。听起来不像你想走的路，但至少你不会遇到这个特定的问题。