Невозможно отфильтровать столбцы CSV, хранящиеся в кадре данных в Spark 2.2.0

Question

Я читаю файл CSV с моего локального компьютера, используя spark и scala, и сохраняю данные в фрейм данных (называемый df).Мне нужно выбрать только несколько выбранных столбцов с новыми именами псевдонимов из df и сохранить их в новом фрейме данных newDf.Я пытался сделать то же самое, но я получаю сообщение об ошибке ниже.

main" org.apache.spark.sql.AnalysisException: cannot resolve '`history_temp.time`' given input columns: [history_temp.time, history_temp.poc]

Ниже приведен код, написанный для чтения файла CSV с моего локального компьютера.

import org.apache.spark.sql.SparkSession

object DataLoadConversion {


  def main(args: Array[String]): Unit = {

    System.setProperty("spark.sql.warehouse.dir", "file:///C:/spark-warehouse")
    val spark = SparkSession.builder().master("local").appName("DataConversion").getOrCreate()

    val df = spark.read.format("com.databricks.spark.csv")
      .option("quote", "\"")
      .option("escape", "\"")
      .option("delimiter", ",")
      .option("header", "true")
      .option("mode", "FAILFAST")
      .option("inferSchema","true")
      .load("file:///C:/Users/an/Desktop/ct_temp.csv")

    df.show(5)   // Till this code is working fine

    val newDf = df.select("history_temp.time","history_temp.poc")

Ниже приведеныкод, который я пробовал, но не работает.

//  val newDf = df.select($"history_temp.time",$"history_temp.poc")
//  val newDf = df.select("history_temp.time","history_temp.poc")
//  val newDf = df.select( df("history_temp.time").as("TIME"))
//  val newDf = df.select(df.col("history_temp.time"))

//  df.select(df.col("*"))    // This is working

    newDf.show(10)
  }
}

Answer 1

судя по всему. формат имени вашего столбца является проблемой здесь. Я предполагаю, что они просто обычный stringType, но когда у вас есть что-то вроде history_temp.time spark думает, что это столбец с массивом. что не так. Я бы переименовал все столбцы и заменил "." до "". тогда вы можете запустить тот же выбор, и он должен работать. Вы можете использовать foldleft, чтобы переместить все "." с "", как показано ниже.

val replacedDF = df.columns.foldleft(df){ (newdf, colname)=>
newdf.withColumnRenamed (colname, colname.replace(".","_"))
}

Сделав это, вы можете выбрать из замененного DF ниже

val newDf= replacedDf.select("history_temp_time","history_temp_poc")

Дайте мне знать, как это работает для вас.