Spark-SQL：如何将 TSV 或 CSV 文件读入数据框并应用自定义模式？

Posted 2023-04-15

【中文标题】Spark-SQL：如何将 TSV 或 CSV 文件读入数据框并应用自定义模式？

【英文标题】：Spark-SQL : How to read a TSV or CSV file into dataframe and apply a custom schema?

【发布时间】：2017-04-20 00:06:53

【问题描述】：

我在处理制表符分隔值 (TSV) 和逗号分隔值 (CSV) 文件时使用的是 Spark 2.0。我想将数据加载到 Spark-SQL 数据帧中，我想在读取文件时完全控制模式。我不希望 Spark 从文件中的数据中猜测架构。

我如何将 TSV 或 CSV 文件加载到 Spark SQL 数据帧中并向它们应用架构？

【参考方案1】：

以下是加载制表符分隔值 (TSV) 文件并应用架构的完整 Spark 2.0 示例。

我以Iris data set in TSV format from UAH.edu 为例。以下是该文件的前几行：

Type    PW      PL      SW      SL
0       2       14      33      50
1       24      56      31      67
1       23      51      31      69
0       2       10      36      46
1       20      52      30      65

要强制实施模式，您可以使用以下两种方法之一以编程方式构建它：

A.使用StructType 创建架构：

import org.apache.spark.sql.types._

var irisSchema = StructType(Array(
    StructField("Type",         IntegerType, true),
    StructField("PetalWidth",   IntegerType, true),
    StructField("PetalLength",  IntegerType, true),
    StructField("SepalWidth",   IntegerType, true),
    StructField("SepalLength",  IntegerType, true)
    ))

B.或者，使用case class 和Encoders 创建架构（这种方法不那么冗长）：

import org.apache.spark.sql.Encoders

case class IrisSchema(Type: Int, PetalWidth: Int, PetalLength: Int, 
                      SepalWidth: Int, SepalLength: Int)

var irisSchema = Encoders.product[IrisSchema].schema

创建架构后，您可以使用spark.read 读取 TSV 文件。请注意，您实际上也可以读取逗号分隔值 (CSV) 文件或任何分隔文件，只要您正确设置了 option("delimiter", d) 选项。此外，如果您有一个带有标题行的数据文件，请务必设置option("header", "true")。

下面是完整的最终代码：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Encoders

val spark = SparkSession.builder().getOrCreate()

case class IrisSchema(Type: Int, PetalWidth: Int, PetalLength: Int,
                      SepalWidth: Int, SepalLength: Int)

var irisSchema = Encoders.product[IrisSchema].schema

var irisDf = spark.read.format("csv").     // Use "csv" regardless of TSV or CSV.
                option("header", "true").  // Does the file have a header line?
                option("delimiter", "\t"). // Set delimiter to tab or comma.
                schema(irisSchema).        // Schema that was built above.
                load("iris.tsv")

irisDf.show(5)

这是输出：

scala> irisDf.show(5)
+----+----------+-----------+----------+-----------+
|Type|PetalWidth|PetalLength|SepalWidth|SepalLength|
+----+----------+-----------+----------+-----------+
|   0|         2|         14|        33|         50|
|   1|        24|         56|        31|         67|
|   1|        23|         51|        31|         69|
|   0|         2|         10|        36|         46|
|   1|        20|         52|        30|         65|
+----+----------+-----------+----------+-----------+
only showing top 5 rows