Struct类型字段对比异常求助：数组正常结构体未展开子字段

我有两个包含Array和Struct类型字段的数据集，需要实现字段级别的差异对比。topLevelReport负责生成顶级字段（比如name）的差异，当差异值大于0时，lowLevelReport会通过curateColumns方法根据数据类型展开嵌套列并计算差异。目前代码对Array类型字段处理正常，但对Struct类型字段存在问题：例如名为name的Struct字段，仅能生成该字段自身的差异值，无法展开其下的brand和keywords子字段进行差异计算。推测是explodeStruct方法的字段映射逻辑，以及selectNestedColumn的匹配逻辑存在问题，导致子字段无法被lowLevelReport识别。

数据集Schema #

name: struct (nullable = true)  
|  |-- brand: string (nullable = true)
|  |-- keywords: string (nullable = true)

现有代码 #

override def validate(df1: DataFrame,
                      df2: DataFrame): Dataset[DiffReportModel] = {

    val columns = curateColumns(df1)
    val topLevelReport: Seq[DiffReportModel] = columns.
        filter(!_.toString.contains(".")).
        map(column => calculateColumnPercentageGap(df1, df2, df1.count, column))
    
    val lowLevelReportFieldNames = topLevelReport.filter(_.anyMistmatch()).map(_.attribute)
   
    val lowLevelReport: Seq[DiffReportModel] = columns.filter(_.toString.contains(".")).
        filter(column => selectNestedColumn(lowLevelReportFieldNames, column)).
        map(column => calculateColumnPercentageGap(df1, df2, df1.count, column))

    (topLevelReport ++ lowLevelReport).toDS() 
}

def explodeArray(df: DataFrame, c: StructField): Seq[Column] = {
    Try(
        df.select(explode(col(c.name)) as "parent")
            .schema.filter(c => c.name == "parent")
            .flatMap(_.dataType.asInstanceOf[StructType].fields)
            .map(f => sort_array(col(c.name)).getField(f.name).as(s"${c.name}.${f.name}"))
    ).getOrElse(Seq()) 
}

def selectNestedColumn(lowLevelReportFieldNames: Seq[String], column:
Column) = {
    lowLevelReportFieldNames.contains(
        column.toString.split(',').head.split('(').last) 
}

def explodeStruct(st: StructType, c: StructField): Seq[Column] = {
    Seq(col(c.name)) ++ st.fields.map(f => col(s"${c.name}.${f.name}")) 
}


def sortAndExplodeArray(df: DataFrame, c: StructField): Seq[Column] =
{
    Seq(sort_array(col(c.name)).as(c.name)) ++ explodeArray(df, c) 
}

/**
 * Return a list of [[Column]]
 * @param schema
 * @return
 */

def curateColumns(df: DataFrame): Seq[Column] = {
    df.schema.flatMap(c => c.dataType match {
        case st: StructType => explodeStruct(st, c)
        case _: ArrayType => sortAndExplodeArray(df, c)         
        case _ => Seq(col(c.name))
    }) 
}

问题根源与修复方案 #

1. 问题分析

• Struct子字段无显式别名：explodeStruct生成的子字段（如name.brand）没有设置别名，虽然Spark内部能识别字段路径，但后续匹配逻辑无法正确提取父字段。
• 子字段匹配逻辑错误：selectNestedColumn通过字符串拆分提取父字段的方式不适用于Struct子字段，导致无法匹配到对应的顶级字段（如从name.brand中提取name）。

2. 修复代码

修正explodeStruct（统一子字段别名格式）

def explodeStruct(st: StructType, c: StructField): Seq[Column] = {
    // 保留顶级Struct字段，为子字段设置parent.child格式的别名，和Array字段处理逻辑对齐
    Seq(col(c.name)) ++ st.fields.map(f => col(s"${c.name}.${f.name}").as(s"${c.name}.${f.name}"))
}

修正selectNestedColumn的匹配逻辑

def selectNestedColumn(lowLevelReportFieldNames: Seq[String], column: Column): Boolean = {
    // 提取子字段的别名（或原始路径），拆分出顶级父字段名
    val columnFullName = column.toString.split(" as ").last.replace("`", "")
    val parentField = columnFullName.split('.').head
    lowLevelReportFieldNames.contains(parentField)
}

修复后效果 #

修复后，curateColumns会生成：

• 顶级字段：name
• 带别名的子字段：name.brand、name.keywords

当topLevelReport中name的差异大于0时，lowLevelReportFieldNames会包含name，selectNestedColumn能正确识别子字段的父字段，从而将name.brand和name.keywords纳入差异计算。

内容的提问来源于stack exchange，提问作者Sakshi Trivedi