Apache Spark-将CSV数据加载到数据集的通用方法

Question

我想用三个输入参数编写通用方法：

1. filePath-字符串
2. 模式-？
3. 案例类

因此，我的想法是编写这样的方法：

def load_sms_ds(filePath: String, schemaInfo: ?, cc: ?) = {
  val ds = spark.read
    .format("csv")
    .option("header", "true")
    .schema(?)
    .option("delimiter",",")
    .option("dateFormat", "yyyy-MM-dd HH:mm:ss.SSS")
    .load(schemaInfo)
    .as[?]

   ds
}

并根据输入参数返回数据集。我不确定参数schemaInfo和cc应该是哪种类型？

Answer 1

首先，我建议阅读spark sql编程指南。这包含一些我认为通常会在您学习火花时对您有所帮助的东西。

让我们使用案例类定义模式来完成读取csv文件的过程。

首先添加此示例所需的变量导入：

import java.io.{File, PrintWriter} // for reading / writing the example data

import org.apache.spark.sql.types.{StringType, StructField} // to define the schema
import org.apache.spark.sql.catalyst.ScalaReflection // used to generate the schema from a case class

import scala.reflect.runtime.universe.TypeTag // used to provide type information of the case class at runtime
import org.apache.spark.sql.Dataset, SparkSession}
import org.apache.spark.sql.Encoder // Used by spark to generate the schema 

定义一个案例类，可以在这里找到不同的类型：

case class Example(
    stringField : String,
    intField : Int,
    doubleField : Double
)

给定案例类类型作为参数，添加用于提取架构的方法（StructType）：

// T : TypeTag means that an implicit value of type TypeTag[T] must be available at the method call site. Scala will automatically generate this for you. See [here][3] for further details. 
def schemaOf[T: TypeTag]: StructType = { 
    ScalaReflection
        .schemaFor[T] // this method requires a TypeTag for T
        .dataType
        .asInstanceOf[StructType] // cast it to a StructType, what spark requires as its Schema
}

Defnie一个方法，该方法从具有使用case类定义的架构的路径中读取csv文件：

// The implicit Encoder is needed by the `.at` method in order to create the Dataset[T]. The TypeTag is required by the schemaOf[T] call.
def readCSV[T : Encoder : TypeTag](
    filePath: String
)(implicit spark : SparkSession) : Dataset[T]= {
    spark.read
        .option("header", "true")
        .option("dateFormat", "yyyy-MM-dd HH:mm:ss.SSS")
        .schema(schemaOf[T])
        .csv(filePath) // spark provides this more explicit call to read from a csv file by default it uses comma and the separator but this can be changes.
        .as[T]
}

创建一个sparkSession：

implicit val spark = SparkSession.builder().master("local").getOrCreate()

将一些示例数据写入临时文件：

val data =
    s"""|stringField,intField,doubleField
        |hello,1,1.0
        |world,2,2.0
    |""".stripMargin
val file = File.createTempFile("test",".csv")
val pw = new PrintWriter(file)
pw.write(data)
pw.close()

调用此方法的示例：

import spark.implicits._ // so that an implicit Encoder gets pulled in for the case class
val df = readCSV[Example](file.getPath)
df.show()