Создание фрейма данных и приведение столбцов с помощью inferSchema из csv намного медленнее, чем создание фрейма с использованием withColumn - PullRequest
Новая
       99

Создание фрейма данных и приведение столбцов с помощью inferSchema из csv намного медленнее, чем создание фрейма с использованием withColumn

1 голос
/ 01 мая 2020

Я пытаюсь создать фрейм данных в Spark, читая csv, проблема в том, что если я ничего не сделаю, в частности, фрейм данных будет иметь каждый тип столбца в виде строки: 

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 секунд. Что дает?

Ответы [ 2 ]

2 голосов
/ 01 мая 2020

Ответ на ваш вопрос: когда вы указываете .option("inferSchema", "true"), он должен динамически читать весь файл, так как вы не указываете процент. Это занимает некоторое время. Это часто не делается с большими файлами.

1 голос
/ 01 мая 2020

Может быть, это может помочь. Ссылаясь на 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)
...