在将 csv 文件作为数据框读取时提供架构答案

【问题标题】：Provide schema while reading csv file as a dataframe在将 csv 文件作为数据框读取时提供架构
【发布时间】：2017-02-16 23:51:44
【问题描述】：

我正在尝试将 csv 文件读入数据框。我知道我的数据框的架构应该是什么，因为我知道我的 csv 文件。另外我正在使用 spark csv 包来读取文件。我试图指定如下架构。

val pagecount = sqlContext.read.format("csv")
  .option("delimiter"," ").option("quote","")
  .option("schema","project: string ,article: string ,requests: integer ,bytes_served: long")
  .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

但是当我检查我创建的数据框的架构时，它似乎采用了自己的架构。我做错什么了吗？如何让 spark 获取我提到的架构？

> pagecount.printSchema
root
|-- _c0: string (nullable = true)
|-- _c1: string (nullable = true)
|-- _c2: string (nullable = true)
|-- _c3: string (nullable = true)

【问题讨论】：

您使用的是哪个版本的 spark ？

标签： scala apache-spark dataframe apache-spark-sql spark-csv

【解决方案1】：

试试下面的代码，你不需要指定架构。当您将 inferSchema 设为 true 时，它应该从您的 csv 文件中获取它。

val pagecount = sqlContext.read.format("csv")
  .option("delimiter"," ").option("quote","")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

如果你想手动指定架构，你可以这样做：

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

val customSchema = StructType(Array(
  StructField("project", StringType, true),
  StructField("article", StringType, true),
  StructField("requests", IntegerType, true),
  StructField("bytes_served", DoubleType, true))
)

val pagecount = sqlContext.read.format("csv")
  .option("delimiter"," ").option("quote","")
  .option("header", "true")
  .schema(customSchema)
  .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

【讨论】：

我尝试执行代码，但它给了我以下错误。 val customSchema = StructType(Array( StructField("project", StringType, true), StructField("article", StringType, true), StructField("requests", IntegerType, true), StructField("bytes_served", DoubleType, true) )) :30: error: not found: value StructType val customSchema = StructType(Array(
理论上我知道我们可以提及架构，但我不知道如何在语法方面提及架构，有什么帮助我可以查找的吗？我参考了官方文档，它没有提到这个案例，也没有太多的例子
这并没有回答问题，尽管它很有帮助。他如何提供架构？
当我尝试 option("inferSchema", "true") 时，我得到了 java.lang.UnsupportedOperationException: Schema inference is not supported for format: csv. Please specify the schema. 这是一个 readStream，所以也许这就是区别。

【解决方案2】：

我在分析中使用了 Arunakiran Nulu 提供的解决方案（请参阅代码）。尽管它能够为列分配正确的类型，但返回的所有值都是null。以前，我尝试过选项.option("inferSchema", "true")，它会在数据框中返回正确的值（尽管类型不同）。

val customSchema = StructType(Array(
    StructField("numicu", StringType, true),
    StructField("fecha_solicitud", TimestampType, true),
    StructField("codtecnica", StringType, true),
    StructField("tecnica", StringType, true),
    StructField("finexploracion", TimestampType, true),
    StructField("ultimavalidacioninforme", TimestampType, true),
    StructField("validador", StringType, true)))

val df_explo = spark.read
        .format("csv")
        .option("header", "true")
        .option("delimiter", "\t")
        .option("timestampFormat", "yyyy/MM/dd HH:mm:ss") 
        .schema(customSchema)
        .load(filename)

结果

root


|-- numicu: string (nullable = true)
 |-- fecha_solicitud: timestamp (nullable = true)
 |-- codtecnica: string (nullable = true)
 |-- tecnica: string (nullable = true)
 |-- finexploracion: timestamp (nullable = true)
 |-- ultimavalidacioninforme: timestamp (nullable = true)
 |-- validador: string (nullable = true)

表格是：

|numicu|fecha_solicitud|codtecnica|tecnica|finexploracion|ultimavalidacioninforme|validador|
+------+---------------+----------+-------+--------------+-----------------------+---------+
|  null|           null|      null|   null|          null|                   null|     null|
|  null|           null|      null|   null|          null|                   null|     null|
|  null|           null|      null|   null|          null|                   null|     null|
|  null|           null|      null|   null|          null|                   null|     null|

【讨论】：

看起来 .option("timestampFormat", "yyyy/mm/dd HH:mm:ss") 应该是 .option("timestampFormat", "yyyy/MM/dd HH:mm:ss")。 [注意月份的大写MM] 否则它将把月份数字解释为时间戳的分钟。
是的！你说的对！我没有注意到。我会编辑我的答案。谢谢
如果您的 DateType 列可能包含“null”值，请设置.option("nullValue", "null")，否则它会认为整行都包含 null 值。
我遇到了同样的问题 - 你有没有在 scala 中得到这个工作？

【解决方案3】：

对于那些有兴趣在 Python 中执行此操作的人，这里有一个工作版本。

customSchema = StructType([
    StructField("IDGC", StringType(), True),        
    StructField("SEARCHNAME", StringType(), True),
    StructField("PRICE", DoubleType(), True)
])
productDF = spark.read.load('/home/ForTesting/testProduct.csv', format="csv", header="true", sep='|', schema=customSchema)

testProduct.csv
ID|SEARCHNAME|PRICE
6607|EFKTON75LIN|890.88
6612|EFKTON100HEN|55.66

希望这会有所帮助。

【讨论】：

【解决方案4】：

感谢@Nulu 的回答，它适用于 pyspark，只需最少的调整

from pyspark.sql.types import LongType, StringType, StructField, StructType, BooleanType, ArrayType, IntegerType

customSchema = StructType(Array(
    StructField("project", StringType, true),
    StructField("article", StringType, true),
    StructField("requests", IntegerType, true),
    StructField("bytes_served", DoubleType, true)))

pagecount = sc.read.format("com.databricks.spark.csv")
         .option("delimiter"," ")
         .option("quote","")
         .option("header", "false")
         .schema(customSchema)
         .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

【讨论】：

以上代码在 pyspark 中不起作用。对我有用的是：>>> customSchema = StructType().add("MovieID", IntegerType(), True).add("Title", StringType(), True).add("Genres", StringType() , True) >>> df = sqlContext.read.format("csv").option("delimiter",",").option("header", "true").schema(customSchema).load("movies .csv") >>> df DataFrame[MovieID: int, Title: string, Genres: string] >>>
数组不是 PySpark 类型。您应该使用 Python 数组 [] 而不是“数组”。并在 Python 中使用“True”而不是“true”。

【解决方案5】：

之前的解决方案都使用了自定义的 StructType。

使用 spark-sql 2.4.5（scala 版本 2.12.10），现在可以使用 schema 函数将架构指定为字符串

import org.apache.spark.sql.SparkSession;

val sparkSession = SparkSession.builder()
            .appName("sample-app")
            .master("local[2]")
            .getOrCreate();

val pageCount = sparkSession.read
  .format("csv")
  .option("delimiter","|")
  .option("quote","")
  .schema("project string ,article string ,requests integer ,bytes_served long")
  .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

【讨论】：

【解决方案6】：

模式定义为简单字符串

以防万一有人对带有 date 和 时间戳 的简单字符串的架构定义感兴趣

从终端或 shell 创建数据文件

echo " 
2019-07-02 22:11:11.000999, 01/01/2019, Suresh, abc  
2019-01-02 22:11:11.000001, 01/01/2020, Aadi, xyz 
" > data.csv

将架构定义为字符串

    user_schema = 'timesta TIMESTAMP,date DATE,first_name STRING , last_name STRING'

读取数据

    df = spark.read.csv(path='data.csv', schema = user_schema, sep=',', dateFormat='MM/dd/yyyy',timestampFormat='yyyy-MM-dd HH:mm:ss.SSSSSS')

    df.show(10, False)

    +-----------------------+----------+----------+---------+
    |timesta                |date      |first_name|last_name|
    +-----------------------+----------+----------+---------+
    |2019-07-02 22:11:11.999|2019-01-01| Suresh   | abc     |
    |2019-01-02 22:11:11.001|2020-01-01| Aadi     | xyz     |
    +-----------------------+----------+----------+---------+

请注意，明确定义架构而不是让 spark 推断架构也可以提高 spark 读取性能。

【讨论】：

val user_schema = "timesta TIMESTAMP,date DATE,first_name STRING , last_name STRING" val mydf1 = spark.read.schema(user_schema).option("dateFormat","MM/dd/yyyy") .option("timestampFormat","yyyy-MM-dd HH:mm:ss.SSSSSS").csv("data.csv")

【解决方案7】：

以下是使用自定义架构的方法，一个完整的演示：

$> 外壳代码，

echo "
Slingo, iOS 
Slingo, Android
" > game.csv

Scala 代码：

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

val customSchema = StructType(Array(
  StructField("game_id", StringType, true),
  StructField("os_id", StringType, true)
))

val csv_df = spark.read.format("csv").schema(customSchema).load("game.csv")
csv_df.show 

csv_df.orderBy(asc("game_id"), desc("os_id")).show
csv_df.createOrReplaceTempView("game_view")
val sort_df = sql("select * from game_view order by game_id, os_id desc")
sort_df.show

【讨论】：

mujiang.blogspot.ca/2018/01/…

【解决方案8】：

你也可以通过使用 sparkSession 和隐式来做到这一点

import sparkSession.implicits._
val pagecount:DataFrame = sparkSession.read
.option("delimiter"," ")
.option("quote","")
.option("inferSchema","true")
.csv("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")
.toDF("project","article","requests","bytes_served")

【讨论】：

【解决方案9】：

如果您的 spark 版本是 3.0.1，您可以使用以下 Scala 脚本：

val df = spark.read.format("csv").option("delimiter",",").option("header",true).load("file:///LOCAL_CSV_FILE_PATH")

但是这样一来，所有的数据类型都会被设置为String。

【讨论】：

【解决方案10】：

这是我们可以在加载 CSV 时将列名传递给数据框的选项之一。

import pandas
    names = ['sepal-length', 'sepal-width', 'petal-length', 'petal-width', 'class']
    dataset = pandas.read_csv("C:/Users/NS00606317/Downloads/Iris.csv", names=names, header=0)
print(dataset.head(10))

输出

    sepal-length  sepal-width  petal-length  petal-width        class
1            5.1          3.5           1.4          0.2  Iris-setosa
2            4.9          3.0           1.4          0.2  Iris-setosa
3            4.7          3.2           1.3          0.2  Iris-setosa
4            4.6          3.1           1.5          0.2  Iris-setosa
5            5.0          3.6           1.4          0.2  Iris-setosa
6            5.4          3.9           1.7          0.4  Iris-setosa
7            4.6          3.4           1.4          0.3  Iris-setosa
8            5.0          3.4           1.5          0.2  Iris-setosa
9            4.4          2.9           1.4          0.2  Iris-setosa
10           4.9          3.1           1.5          0.1  Iris-setosa

【讨论】：

很好的答案，但问题是指 Spark DataFrame，而不是 Pandas 的那个

【解决方案11】：

// import Library
import java.io.StringReader ;

import au.com.bytecode.opencsv.CSVReader

//filename

var train_csv = "/Path/train.csv";

//read as text file

val train_rdd = sc.textFile(train_csv)   

//use string reader to convert in proper format

var full_train_data  = train_rdd.map{line =>  var csvReader = new CSVReader(new StringReader(line)) ; csvReader.readNext();  }   

//declares  types

type s = String

// declare case class for schema

case class trainSchema (Loan_ID :s ,Gender :s, Married :s, Dependents :s,Education :s,Self_Employed :s,ApplicantIncome :s,CoapplicantIncome :s,
    LoanAmount :s,Loan_Amount_Term :s, Credit_History :s, Property_Area :s,Loan_Status :s)

//create DF RDD with custom schema 

var full_train_data_with_schema = full_train_data.mapPartitionsWithIndex{(idx,itr)=> if (idx==0) itr.drop(1); 
                     itr.toList.map(x=> trainSchema(x(0),x(1),x(2),x(3),x(4),x(5),x(6),x(7),x(8),x(9),x(10),x(11),x(12))).iterator }.toDF

【讨论】：

【解决方案12】：

在 pyspark 2.4 以后，您可以简单地使用header 参数来设置正确的标头：

data = spark.read.csv('data.csv', header=True)

同样，如果使用 scala，您也可以使用 header 参数。

【讨论】：

【解决方案13】：

我的解决方案是：

import org.apache.spark.sql.types._
  val spark = org.apache.spark.sql.SparkSession.builder.
  master("local[*]").
  appName("Spark CSV Reader").
  getOrCreate()

val movie_rating_schema = StructType(Array(
  StructField("UserID", IntegerType, true),
  StructField("MovieID", IntegerType, true),
  StructField("Rating", DoubleType, true),
  StructField("Timestamp", TimestampType, true)))

val df_ratings: DataFrame = spark.read.format("csv").
  option("header", "true").
  option("mode", "DROPMALFORMED").
  option("delimiter", ",").
  //option("inferSchema", "true").
  option("nullValue", "null").
  schema(movie_rating_schema).
  load(args(0)) //"file:///home/hadoop/spark-workspace/data/ml-20m/ratings.csv"

val movie_avg_scores = df_ratings.rdd.map(_.toString()).
  map(line => {
    // drop "[", "]" and then split the str 
    val fileds = line.substring(1, line.length() - 1).split(",")
    //extract (movie id, average rating)
    (fileds(1).toInt, fileds(2).toDouble)
  }).
  groupByKey().
  map(data => {
    val avg: Double = data._2.sum / data._2.size
    (data._1, avg)
  })

【讨论】：