斯卡拉 Spark : Performance issue renaming huge number of columns

Question

为了能够在不转义 的情况下使用我的 DataFrame 的列名。 我需要一个函数来“验证”所有列名 - 但我尝试过的方法都没有及时的工作(我在 5 分钟后中止)。

我正在尝试我的算法的数据集是 golub 数据集(获取它 here )。这是一个包含 7200 列的 2.2MB CSV 文件。重命名所有列应该是几秒钟的事情

读取 CSV 的代码

var dfGolub = spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("golub_merged.csv")
    .drop("_c0") // drop the first column
    .repartition(numOfCores)

尝试重命名列:

 def validifyColumnnames1(df : DataFrame) : DataFrame = {
     import org.apache.spark.sql.functions.col
     val cols = df.columns
     val colsRenamed = cols.map(name => col(name).as(name.replaceAll("\\.","")))
     df.select(colsRenamed : _*)
 }


def validifyColumnnames2[T](df : Dataset[T]) : DataFrame = {
    val newColumnNames = ArrayBuffer[String]()
    for(oldCol <- df.columns) {
        newColumnNames +=  oldCol.replaceAll("\\.","")
    }
    df.toDF(newColumnNames : _*)
}

def validifyColumnnames3(df : DataFrame) : DataFrame = {
    var newDf = df
    for(col <- df.columns){
        newDf = newDf.withColumnRenamed(col,col.replaceAll("\\.",""))
    }
    newDf
}

知道是什么导致了这个性能问题吗？

设置:我在一台 16 核 * 2 线程和 96GB RAM 的机器上以 local[24] 模式在 Ubuntu 16.04 上运行 Spark 2.1.0

Answer 1

假设您知道类型，您可以简单地创建模式而不是推断它(推断模式会降低性能并且对于 csv 甚至可能是错误的)。

为简单起见，假设您有如下文件 example.csv:

A.B, A.C, A.D
a,3,1

你可以这样做:

val scehma = StructType(Seq(StructField("A_B",StringType),StructField("A_C", IntegerType), StructField("AD", IntegerType)))
val df = spark.read.option("header","true").schema(scehma).csv("example.csv")
df.show()

+---+---+---+
|A_B|A_C| AD|
+---+---+---+
|  a|  3|  1|
+---+---+---+

如果您事先不知道信息，您可以像以前一样使用推断架构，然后您可以使用数据框生成架构:

val fields = for {
  x <- df.schema
} yield StructField(x.name.replaceAll("\\.",""), x.dataType, x.nullable)
val schema = StructType(fields)

然后像以前一样使用该模式重新读取数据框