动态和可配置地更改几种Spark DataFrame列类型

Question

我是Spark和Scala的新手。

我们有一个外部数据源提供给我们JSON。此JSON将所有值括起来，包括数字和布尔值字段。所以当我把它放到我的DataFrame中时，所有的列都是字符串。最终目标是将这些JSON记录转换为类型正确的Parquet文件。

大约有100个字段，我需要将其中几个类型从string更改为int、boolean或bigint (long)。此外，我们处理的每个DataFrame都只有这些字段的一个子集，而不是所有字段。因此，我需要能够处理给定DataFrame的列子集，将每列与已知的列类型列表进行比较，并根据DataFrame中出现的列将某些列从字符串转换为整型、大整型和布尔型。

最后，我需要列类型列表是可配置的，因为我们将来会有新的列，并且可能想要删除或更改旧的列。

所以，这就是我到目前为止所知道的：

// first I convert to all lower case for column names
val df = dfIn.toDF(dfIn.columns map(_.toLowerCase): _*)

// Big mapping to change types
// TODO how would I make this configurable?
// I'd like to drive this list from an external config file.
val dfOut = df.select(
   df.columns.map {

     ///// Boolean
     case a @ "a" => df(a).cast(BooleanType).as(a)
     case b @ "b" => df(b).cast(BooleanType).as(b)

     ///// Integer
     case i @ "i" => df(i).cast(IntegerType).as(i)
     case j @ "j" => df(j).cast(IntegerType).as(j)


     // Bigint to Double
     case x @ "x" => df(x).cast(DoubleType).as(x)
     case y @ "y" => df(y).cast(DoubleType).as(y)

     case other         => df(other)
   }: _*
)

这是一种将这些数据转换为我想要的Scala类型的有效方法吗？

我可以使用一些关于如何从外部'config‘文件驱动它的建议，在这个文件中，我可以定义列类型。