如何在不使用Scala案例类的情况下指定CSV文件的架构？

Question

我正在将CSV文件加载到DataFrame中,如下所示.

val conf=new SparkConf().setAppName("dataframes").setMaster("local")
val sc=new SparkContext(conf)
val spark=SparkSession.builder().getOrCreate()
import spark.implicits._

val df = spark.
  read.  
  format("org.apache.spark.csv").
  option("header", true).
  csv("/home/cloudera/Book1.csv")
scala> df.printSchema()
root
 |-- name: string (nullable = true)
 |-- address: string (nullable = true)
 |-- age: string (nullable = true)

如何将age列更改为类型Int？

Answer 1

鉴于val spark=SparkSession.builder().getOrCreate()我猜你正在使用Spark 2.x.

首先,请注意Spark 2.x具有CSV格式的原生支持,因此不需要通过其长名称来指定格式,即org.apache.spark.csv仅仅csv.

spark.read.format("csv")...

由于您使用csv运算符,因此隐含了CSV格式,因此您可以跳过/删除format("csv").

// note that I removed format("csv")
spark.read.option("header", true).csv("/home/cloudera/Book1.csv")

有了这个,你有很多选择,但我强烈建议使用一个案例类...只是架构.如果您对如何在Spark 2.0中执行此操作感到好奇,请参阅上一个解决方案.

演员

你可以使用强制算子.

scala> Seq("1").toDF("str").withColumn("num", 'str cast "int").printSchema
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = true)

使用StructType

您还可以使用自己的手工模式与StructType和StructField,如下所示:

import org.apache.spark.sql.types._    
val schema = StructType(
  StructField("str", StringType, true) :: 
  StructField("num", IntegerType, true) :: Nil)

scala> schema.printTreeString
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = true)

val q = spark.
  read.
  option("header", true).
  schema(schema).
  csv("numbers.csv")
scala> q.printSchema
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = true)

架构DSL

我最近发现的很有趣的是所谓的Schema DSL.使用上述模式建立StructType和StructField可重写如下:

import org.apache.spark.sql.types._
val schema = StructType(
  $"str".string ::
  $"num".int :: Nil) 
scala> schema.printTreeString
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = true)

// or even
val schema = new StructType().
  add($"str".string).
  add($"num".int)
scala> schema.printTreeString
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = true)

编码器

编码器很容易使用,很难相信你不能想要它们,甚至只是在没有处理的情况下构建一个模式StructType,StructField而且DataType.

// Define a business object that describes your dataset
case class MyRecord(str: String, num: Int)

// Use Encoders object to create a schema off the business object
import org.apache.spark.sql.Encoders    
val schema = Encoders.product[MyRecord].schema
scala> schema.printTreeString
root
 |-- str: string (nullable = true)
 |-- num: integer (nullable = false)

Answer 2

可以inferSchema选择通过以下方式自动识别变量的类型:

val df=spark.read
  .format("org.apache.spark.csv")
  .option("header", true)
  .option("inferSchema", true) // <-- HERE
  .csv("/home/cloudera/Book1.csv")

spark-csv最初是databricks的外部库,但包含在spark 2.0版本的核心火花中.您可以参考库的github页面上的文档来查找可用选项.