在 apache spark 中使用模式解析文件答案

【问题标题】：parse file with schema in apache spark在 apache spark 中使用模式解析文件
【发布时间】：2017-09-11 10:20:17
【问题描述】：

下面是我的 spark/SCALA 程序，用于读取我的源文件。（CSV 文件）

val csv = spark.read
  .format("com.databricks.spark.csv")
  .option("header", "true") //reading the headers
 // .option("mode", "DROPMALFORMED")
  .option("inferSchema", "true")

  .load("C:\\TestFiles\\SAP_ENT_INVBAL.csv"); //.csv("csv/file/path") //spark 2.0 api


csv.show()



csv.printSchema()
csv.show()

}

输出包含文件头，但对于我的处理，我需要不同的命名约定而不是文件头。

我已经尝试了几个选项并且效果很好。

重命名数据框列
使用 add(StructField 函数

但我想让我的代码通用。只需在读取文件时传递架构文件并根据架构文件创建带有列的数据框。

请帮助解决这个问题。

【问题讨论】：

标签： java scala apache-spark

【解决方案1】：

这里是 spark-csv 文档中关于如何指定自定义模式的示例-

读取数据时可以手动指定schema：

import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

val sqlContext = new SQLContext(sc)       
val customSchema = StructType(Array(
    StructField("year", IntegerType, true),
    StructField("make", StringType, true),
    StructField("model", StringType, true),
    StructField("comment", StringType, true),
    StructField("blank", StringType, true)))

val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true") // Use first line of all files as header
    .schema(customSchema)
    .load("cars.csv")

【讨论】：

这可行，但是我如何将 customSchema 存储在文件中并将其传递给 .schema？

【解决方案2】：

如果您只需要重命名列，可以使用toDF 方法，将列的新名称传递给它，例如

val csv = spark.read.option("header", "true")
  .csv("C:\\TestFiles\\SAP_ENT_INVBAL.csv")
  .toDF("newColAName", "newColBName", "newColCName")

【讨论】：