scala - 结合没有重复的 Spark 模式？

Question

为了处理我拥有的数据，我之前提取了模式，这样当我读取数据集时，我就提供了模式，而不是经历昂贵的推断模式的步骤。

为了构建schema，我需要将几个不同的schema合并到最终的schema中，所以我一直在使用union (++)和distinct方法，但我不断收到 org.apache.spark.sql.AnalysisException: Duplicate column(s) 异常。

例如，假设我们有以下结构中的两个模式:

val schema1 = StructType(StructField("A", StructType(
    StructField("i", StringType, true) :: Nil
    ), true) :: Nil)

val schema2 = StructType(StructField("A", StructType(
    StructField("i", StringType, true) :: Nil
    ), true) :: Nil)

val schema3 = StructType(StructField("A", StructType(
    StructField("i", StringType, true) ::
    StructField("ii", StringType, true) :: Nil
    ), true) :: Nil)

val final_schema = (schema1 ++ schema2 ++ schema3).distinct

println(final_schema)

输出:

StructType(
    StructField(A,StructType(
         StructField(i,StringType,true)),true), 
    StructField(A,StructType(
        StructField(i,StringType,true),    
        StructField(ii,StringType,true)),true))

我知道只有与另一个模式完全匹配的模式结构才会被 distinct 过滤掉。但是我希望结果看起来像这样:

StructType(
    StructField(A,StructType(
        StructField(i,StringType,true),    
        StructField(ii,StringType,true)),true))

其中所有“组合”到一个模式中。 scala documentation中的所有方法我都筛选过了但我似乎找不到解决这个问题的正确方法。有什么想法吗？

编辑:

最终目标是将 final_schema 提供给 sqlContext.read.schema 并使用 read 方法读取 JSON 字符串的 RDD。

Answer 1

尝试这样的事情:

(schema1 ++ schema2 ++ schema3).groupBy(getKey).map(_._2.head)

其中 getKey 是一个函数，它从架构到您要考虑合并的属性(例如列名或子字段的名称)。在 map 函数中，您可以使用 head 或使用一些更精细的函数来保留特定的模式。