用Spark Scala将嵌套数组和StructType的Parquet文件展平

要将嵌套数组和StructType的Parquet文件展平，可以使用Spark Scala中的explode函数和withColumn函数。以下是一个示例代码，演示了如何展平嵌套数组和StructType的Parquet文件：

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions.{col, explode, struct}

object FlattenParquetFile {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = SparkSession.builder()
      .appName("Flatten Parquet File")
      .master("local")
      .getOrCreate()

    // 读取Parquet文件
    val df = spark.read.parquet("path/to/parquet/file")

    // 定义展平函数
    def flattenDataFrame(df: DataFrame): DataFrame = {
      // 获取所有列名
      val columns = df.schema.fields.map(_.name)

      // 遍历所有列
      val flattenedColumns = columns.flatMap { colName =>
        // 获取列的数据类型
        val dataType = df.schema(colName).dataType

        dataType match {
          case st: org.apache.spark.sql.types.StructType =>
            // 如果列的数据类型为StructType，则递归调用展平函数
            val nestedColumns = st.fields.map(field => s"$colName.${field.name}")
            flattenDataFrame(df.select(nestedColumns.map(col): _*))

          case _ =>
            // 如果列的数据类型不是StructType，则返回原始列
            Seq(col(colName))
        }
      }

      // 使用explode函数展平数组列
      val explodedDF = df.withColumn("exploded", explode(col("array_column")))
        .select(flattenedColumns: _*)

      explodedDF
    }

    // 展平Parquet文件
    val flattenedDF = flattenDataFrame(df)

    // 打印展平后的DataFrame
    flattenedDF.show()
  }
}

在上述代码中，首先创建了一个SparkSession对象，并使用spark.read.parquet方法读取Parquet文件。然后定义了一个名为flattenDataFrame的函数，用于展平DataFrame。该函数使用schema属性获取所有列名，并遍历每个列。

对于每个列，首先检查其数据类型。如果数据类型是StructType，则递归调用flattenDataFrame函数，将该列展平为多个列。如果数据类型不是StructType，则保留原始列。

最后，使用explode函数展平数组列，并使用select方法选择展平后的所有列。最终得到展平后的DataFrame。

在主函数中，调用flattenDataFrame函数，将Parquet文件展平为DataFrame，并使用show方法打印展平后的DataFrame。