Spark实现复杂AVRO Schema转CSV的简便解决方案求助

问题背景 #

有一个结构复杂的AVRO文件，Schema片段如下（完整Schema约300行）：

root
 |-- identifier: struct (nullable = true)
 |    |-- domain: string (nullable = true)
 |    |-- id: string (nullable = true)
 |    |-- version: long (nullable = true)
 |-- recordOperator: string (nullable = true)
 |-- payload: struct (nullable = true)
 |    |-- identity: struct (nullable = true)
 |    |    |-- domain: string (nullable = true)
 |    |    |-- id: string (nullable = true)
 |    |    |-- version: long (nullable = true)
 |    |-- principalPaymentPeriodDates: struct (nullable = true)
 |    |    |-- frequency: struct (nullable = true)
 |    |    |    |-- frequency: struct (nullable = true)
 |    |    |    |    |-- issuer: string (nullable = true)
 |    |    |    |    |-- schemeName: string (nullable = true)
 |    |    |    |    |-- value: string (nullable = true)
 |    |    |    |-- count: double (nullable = true)
 |    |    |-- firstRegularPeriodStartDate: string (nullable = true)
 |    |    |-- lastRegularPeriodEndDate: string (nullable = true)
 |    |    |-- startDate: string (nullable = true)
 |    |    |-- endDate: string (nullable = true)
 ..........................

尝试用以下代码转换为CSV时报错：

df.write().option("header",true).format("csv").save("C:\\Users\\my_output.csv");

错误信息：

Exception in thread "main" org.apache.spark.sql.AnalysisException: CSV data source does not support structdomain:string,id:string,version:bigint data type.;

手动展开所有Struct列过于繁琐，需要快速解决方案。

解决方案 #

方法1：递归遍历Schema，自动展开所有嵌套Struct #

编写递归函数遍历DataFrame的Schema，将所有嵌套Struct列展开为扁平化列（用.或_分隔层级），重构DataFrame后再写入CSV。

Scala示例代码

import org.apache.spark.sql.types.{StructType, StructField}
import org.apache.spark.sql.Column
import org.apache.spark.sql.functions.col

def flattenSchema(schema: StructType, prefix: String = null): Array[Column] = {
  schema.fields.flatMap(field => {
    val columnName = if (prefix == null) field.name else s"$prefix.${field.name}"
    field.dataType match {
      case structType: StructType => flattenSchema(structType, columnName)
      case _ => Array(col(columnName).alias(columnName.replace(".", "_"))) // 用下划线替换点，避免部分工具解析问题
    }
  })
}

// 生成扁平化DataFrame
val flattenedDf = df.select(flattenSchema(df.schema): _*)

// 写入CSV
flattenedDf.write.option("header", true).format("csv").save("C:\\Users\\my_output.csv")

Java示例代码

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.Column;
import static org.apache.spark.sql.functions.col;

import java.util.ArrayList;
import java.util.List;

public class FlattenUtils {
    public static List<Column> flattenSchema(StructType schema, String prefix) {
        List<Column> columns = new ArrayList<>();
        for (StructField field : schema.fields()) {
            String colName = (prefix == null) ? field.name() : prefix + "." + field.name();
            if (field.dataType() instanceof StructType) {
                columns.addAll(flattenSchema((StructType) field.dataType(), colName));
            } else {
                columns.add(col(colName).alias(colName.replace(".", "_")));
            }
        }
        return columns;
    }

    // 使用示例
    public static void main(String[] args) {
        // 假设df是读取AVRO后的DataFrame
        List<Column> flattenedCols = flattenSchema(df.schema(), null);
        Dataset<Row> flattenedDf = df.select(flattenCols.toArray(new Column[0]));
        flattenedDf.write().option("header", true).format("csv").save("C:\\Users\\my_output.csv");
    }
}

方法2：手动展开顶层Struct（适用于嵌套层级少的场景） #

如果只有少数几层嵌套，可以直接用select配合*展开顶层Struct，比如：

val partiallyFlattenedDf = df.select("identifier.*", "recordOperator", "payload.identity.*", "payload.principalPaymentPeriodDates.*")

但多层嵌套场景下，递归方法更高效通用。

注意事项 #

• 扁平化后的列名可选择保留.或替换为_，避免部分CSV解析工具出现异常
• 如果存在数组、Map等其他复杂类型，CSV同样不支持，需要额外处理（比如将数组展开为多行，Map转为键值对列）
• 确保所有列类型为CSV支持的类型（字符串、数值、日期等），必要时可通过cast转换类型

内容的提问来源于stack exchange，提问作者Zaid Khan