从 csv 创建数据框并使用 inferSchema 转换列比创建数据框和使用 withColumn 转换要慢得多答案

【问题标题】：Creating a dataframe and casting colums with inferSchema from a csv is way slower than creating one and casting with withColumn从 csv 创建数据框并使用 inferSchema 转换列比创建数据框和使用 withColumn 转换要慢得多
【发布时间】：2020-08-16 04:39:52
【问题描述】：

我正在尝试通过读取 csv 在 Spark 中创建数据框，问题是如果我不做任何特别的事情，数据框的每个列类型都会作为字符串：

root
 |-- ticker: string (nullable = true)
 |-- open: string (nullable = true)
 |-- close: string (nullable = true)
 |-- adj_close: string (nullable = true)
 |-- low: string (nullable = true)
 |-- high: string (nullable = true)
 |-- volume: string (nullable = true)
 |-- date: string (nullable = true)

为了解决这个问题，我将选项“inferSchema”添加为 true，如下所示：

val spark = SparkSession.builder
.appName("Job One")
.master("local")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", spark_events)
.getOrCreate()
import spark.implicits._

val df = spark.read
     .format("csv")
     .option("inferSchema", "true")
     .option("header", "true") 
     .option("mode", "DROPMALFORMED")
     .load(historicalStockPrices)

df.printSchema()

通过这种方式我得到了这个：

root
 |-- ticker: string (nullable = true)
 |-- open: double (nullable = true)
 |-- close: double (nullable = true)
 |-- adj_close: double (nullable = true)
 |-- low: double (nullable = true)
 |-- high: double (nullable = true)
 |-- volume: long (nullable = true)
 |-- date: string (nullable = true)

这正是我想要的，但是添加选项 inferSchema 使得当我不添加它时，这项工作需要 1.4 分钟而不是 6 秒。另一种获取我想要的类型的列的方法是使用 withColumn，如下所示：

val df2 = df
.withColumn("open",df("open").cast("Float"))
.withColumn("close",df("close").cast("Float"))
.withColumn("adj_close",df("adj_close").cast("Float"))
.withColumn("low",df("low").cast("Float"))
.withColumn("high",df("high").cast("Float"))
.withColumn("volume",df("volume").cast("Long"))

df2.printSchema()

这次整个操作的结果又是6秒而已。什么给了？

【问题讨论】：

标签： scala dataframe apache-spark apache-spark-sql

【解决方案1】：

您的问题的答案：当您指定.option("inferSchema", "true") 时，它需要动态读取整个文件，因为您没有指定百分比。这需要一段时间。大文件通常不会这样做。

【讨论】：

【解决方案2】：

也许这会有所帮助。参考this，您为什么不尝试创建自己的StructType 架构，然后在load 之前，您可以使用schema 方法。因此，在阅读 CSV 时，您的代码将如下所示：

//Assuming you've already created your schema

val df = spark.read
   .format("csv")
   .option("header", "true")
   .schema(customSchema) 
   .option("mode", "DROPMALFORMED")
   .load(historicalStockPrices)

【讨论】：