如何将两个具有不同结构类型字段的 spark Dataframe 联合起来？答案

【问题标题】：How to union two spark Dataframe with a field of type struct that can differ?如何将两个具有不同结构类型字段的 spark Dataframe 联合起来？
【发布时间】：2018-07-22 11:55:52
【问题描述】：

我对 Apache Spark 还很陌生，但有时我仍然在苦苦挣扎。我正在尝试导入一个非常复杂的 json 文件并将其展平，然后将其保存在 parquet 文件中。

我的 json 文件是一棵商店树。

{
"id": "store02",
"name": "store name",
"domain": "domain",
"currency": "EUR",
"address1": "Somewhere",
"country": "GER",
"city": "Berlin",
"zipCode": "12345",
"timeZone": "CET",
"accounts" : [
    {
        "field1": "",
        "filed2": "",
        "field3": "",
        "optionnalArray1": [
            {
                "field1": "",
                "field2": ""
            }
        ],
        "optionnalArray2": ["aa", "bb"]
    }
],
"stores": [ .... ]    
}

每个商店都可以有一个字段，该字段是一组帐户。一个帐户有 3 个必填字段和两个可选字段。所以我有一个数据框，其中的字段可以有 3 种不同的类型。

在数据框中导入文件没什么大不了的，但在展平过程中，我可能想对两个具有不同架构的帐户的数据框进行联合，当然我有以下错误：“联合只能是在具有兼容列类型的表上执行"

有没有办法轻松做到这一点？怎样才能顺利导入这样的json文件？

@拉梅什

假设我有两个数据框。第一个是没有账户的商店的数据框。第二个是带有帐户的商店的数据框。帐户是这样定义的结构：

val acquirerStruct = StructType(
    StructField("merchantId", StringType, nullable = true) ::
    StructField("name", StringType, nullable = true) ::
    Nil)

val accountStruct = StructType(
    StructField("acquirers", ArrayType(acquirerStruct), nullable = true) ::
        StructField("applicationCode", StringType, nullable = true) ::
        StructField("channelType", StringType, nullable = true) ::
        StructField("id", StringType, nullable = true) ::
        StructField("terminals", ArrayType(StringType), nullable = true) ::
        Nil)

当我想合并两个数据框时，我为之前的第一个数据框创建了一个列帐户：

df1.withColumn("account", array(lit(null).cast(accountStruct))).union(df2)

如果在 df2 中，所有行都有一个与 accountStruct 结构相同的帐户，则它可以正常工作。但这并不总是正确的。一个账户可能没有终端或收单机构。这在 json 中完全有效。在那种情况下，我有之前提到的错误。

"Union can only be performed on tables with the compatible column types"

【问题讨论】：

你能分享你到目前为止所尝试的吗？ ??
您应该在问题本身中添加以下答案。请这样做:)
好的。完毕。我不知道它是这样工作的。
你不能写一个 if else 语句来检查 df2 中的 account 列吗？
是的，我可以，但我想避免这种情况，并有一个尽可能简单的代码。事实上，我已经有一个几乎相同的解决方法，但我很沮丧不能在一行代码中做这样的事情，我不明白为什么 Spark 似乎完全能够处理这种类型的列并且不允许我这样做。

标签： scala apache-spark struct spark-dataframe

【解决方案1】：

我在 PySpark 中遇到了同样的问题，我通过在读取不兼容的数据帧时提供架构解决了它

import copy
...
schema_to_read = copy.deepcopy(df1.schema)
df2 = sql_context.read.format("json").schema(schema_to_read).load(path)

【讨论】：