link管理

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

CSVSuite.scala

* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements. See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License. You may obtain a copy of the License at

* http://www.apache.org/licenses/LICENSE-2.0

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

package org . apache . spark . sql . execution . datasources . csv

import java . io .{ EOFException , File }

import java . nio . charset .{ Charset , StandardCharsets , UnsupportedCharsetException }

import java . nio . file .{ Files , StandardOpenOption }

import java . sql .{ Date , Timestamp }

import java . text . SimpleDateFormat

import java . time .{ Duration , Instant , LocalDate , LocalDateTime , Period }

import java . util . Locale

import scala . jdk . CollectionConverters . _

import scala . util . Properties

import com . univocity . parsers . common . TextParsingException

import org . apache . commons . lang3 . time . FastDateFormat

import org . apache . hadoop . fs . Path

import org . apache . hadoop . io . SequenceFile . CompressionType

import org . apache . hadoop . io . compress . GzipCodec

import org . apache . logging . log4j . Level

import org . apache . spark .{ SparkConf , SparkException , SparkIllegalArgumentException , SparkRuntimeException , SparkUpgradeException , TestUtils }

import org . apache . spark . sql .{ AnalysisException , Column , DataFrame , Encoders , QueryTest , Row }

import org . apache . spark . sql . catalyst . csv . CSVOptions

import org . apache . spark . sql . catalyst . util .{ DateTimeTestUtils , DateTimeUtils , HadoopCompressionCodec }

import org . apache . spark . sql . execution . datasources . CommonFileDataSourceSuite

import org . apache . spark . sql . internal .{ LegacyBehaviorPolicy , SQLConf }

import org . apache . spark . sql . test . SharedSparkSession

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

abstract class CSVSuite

extends QueryTest

with SharedSparkSession

with TestCsvData

with CommonFileDataSourceSuite {

import testImplicits . _

override protected def dataSourceFormat = " csv "

protected val carsFile = " test-data/cars.csv "

protected val productsFile = " test-data/products.csv "

private val carsMalformedFile = " test-data/cars-malformed.csv "

private val carsFile8859 = " test-data/cars_iso-8859-1.csv "

private val carsTsvFile = " test-data/cars.tsv "

private val carsAltFile = " test-data/cars-alternative.csv "

private val carsMultiCharDelimitedFile = " test-data/cars-multichar-delim.csv "

private val carsMultiCharCrazyDelimitedFile = " test-data/cars-multichar-delim-crazy.csv "

private val carsUnbalancedQuotesFile = " test-data/cars-unbalanced-quotes.csv "

private val carsNullFile = " test-data/cars-null.csv "

private val carsEmptyValueFile = " test-data/cars-empty-value.csv "

private val carsBlankColName = " test-data/cars-blank-column-name.csv "

private val carsCrlf = " test-data/cars-crlf.csv "

private val emptyFile = " test-data/empty.csv "

private val commentsFile = " test-data/comments.csv "

private val disableCommentsFile = " test-data/disable_comments.csv "

private val boolFile = " test-data/bool.csv "

private val decimalFile = " test-data/decimal.csv "

private val simpleSparseFile = " test-data/simple_sparse.csv "

private val numbersFile = " test-data/numbers.csv "

private val datesFile = " test-data/dates.csv "

private val dateInferSchemaFile = " test-data/date-infer-schema.csv "

private val unescapedQuotesFile = " test-data/unescaped-quotes.csv "

private val valueMalformedFile = " test-data/value-malformed.csv "

private val badAfterGoodFile = " test-data/bad_after_good.csv "

private val malformedRowFile = " test-data/malformedRow.csv "

private val charFile = " test-data/char.csv "

/** Verifies data and schema. */

private def verifyCars (

df : DataFrame ,

withHeader : Boolean ,

numCars : Int = 3 ,

numFields : Int = 5 ,

checkHeader : Boolean = true ,

checkValues : Boolean = true ,

checkTypes : Boolean = false ) : Unit = {

val numColumns = numFields

val numRows = if (withHeader) numCars else numCars + 1

// schema

assert(df.schema.fieldNames.length === numColumns)

assert(df.count() === numRows)

if (checkHeader) {

if (withHeader) {

assert(df.schema.fieldNames === Array ( " year " , " make " , " model " , " comment " , " blank " ))

} else {

assert(df.schema.fieldNames === Array ( " _c0 " , " _c1 " , " _c2 " , " _c3 " , " _c4 " ))

}

if (checkValues) {

val yearValues = List ( " 2012 " , " 1997 " , " 2015 " )

val actualYears = if ( ! withHeader) " year " :: yearValues else yearValues

val years = if (withHeader) df.select( " year " ).collect() else df.select( " _c0 " ).collect()

years.zipWithIndex.foreach { case (year, index) =>

if (checkTypes) {

assert(year === Row (actualYears(index).toInt))

} else {

assert(year === Row (actualYears(index)))

}

test( " simple csv test " ) {

val cars = spark

.read

.format( " csv " )

.option( " header " , " false " )

.load(testFile(carsFile))

verifyCars(cars, withHeader = false , checkTypes = false )

}

test( " simple csv test with calling another function to load " ) {

val cars = spark

.read

.option( " header " , " false " )

.csv(testFile(carsFile))

verifyCars(cars, withHeader = false , checkTypes = false )

}

test( " simple csv test with type inference " ) {

val cars = spark

.read

.format( " csv " )

.option( " header " , " true " )

.option( " inferSchema " , " true " )

.load(testFile(carsFile))

verifyCars(cars, withHeader = true , checkTypes = true )

}

test( " simple csv test with string dataset " ) {

val csvDataset = spark.read.text(testFile(carsFile)).as[ String ]

val cars = spark.read

.option( " header " , " true " )

.option( " inferSchema " , " true " )

.csv(csvDataset)

verifyCars(cars, withHeader = true , checkTypes = true )

val carsWithoutHeader = spark.read

.option( " header " , " false " )

.csv(csvDataset)

verifyCars(carsWithoutHeader, withHeader = false , checkTypes = false )

}

test( " test inferring booleans " ) {

val result = spark.read

.format( " csv " )

.option( " header " , " true " )

.option( " inferSchema " , " true " )

.load(testFile(boolFile))

val expectedSchema = StructType ( List (

StructField ( " bool " , BooleanType , nullable = true )))

assert(result.schema === expectedSchema)

}

test( " test inferring decimals " ) {

val result = spark.read

.format( " csv " )

.option( " comment " , " ~ " )

.option( " header " , " true " )

.option( " inferSchema " , " true " )

.load(testFile(decimalFile))

val expectedSchema = StructType ( List (

StructField ( " decimal " , DecimalType ( 20 , 0 ), nullable = true ),

StructField ( " long " , LongType , nullable = true ),

StructField ( " double " , DoubleType , nullable = true )))

assert(result.schema === expectedSchema)

}

test( " test with alternative delimiter and quote " ) {

val cars = spark.read

.format( " csv " )

.options( Map ( " quote " - > " \' " , " delimiter " - > " | " , " header " - > " true " ))

.load(testFile(carsAltFile))

verifyCars(cars, withHeader = true )

}

test( " test with tab delimiter and double quote " ) {

val cars = spark.read

.options( Map ( " quote " - > " \" " , " delimiter " - > """ \t """ , " header " - > " true " ))

.csv(testFile(carsTsvFile))

verifyCars(cars, numFields = 6 , withHeader = true , checkHeader = false )

}

test( " SPARK-24540: test with multiple character delimiter (comma space) " ) {

val cars = spark.read

.options( Map ( " quote " - > " \' " , " delimiter " - > " , " , " header " - > " true " ))

.csv(testFile(carsMultiCharDelimitedFile))

verifyCars(cars, withHeader = true )

}

test( " SPARK-24540: test with multiple (crazy) character delimiter " ) {

val cars = spark.read

.options( Map ( " quote " - > " \' " , " delimiter " - > """ _/- \\ _ """ , " header " - > " true " ))

.csv(testFile(carsMultiCharCrazyDelimitedFile))

verifyCars(cars, withHeader = true )

// check all the other columns, besides year (which is covered by verifyCars)

val otherCols = cars.select( " make " , " model " , " comment " , " blank " ).collect()

val expectedOtherColVals = Seq (

( " Tesla " , " S " , " No comment " , null ),

( " Ford " , " E350 " , " Go get one now they are going fast " , null ),

( " Chevy " , " Volt " , null , null )

)

expectedOtherColVals.zipWithIndex.foreach { case (values, index) =>

val actualRow = otherCols(index)

values match {

case (make, model, comment, blank) =>

assert(make == actualRow.getString( 0 ))

assert(model == actualRow.getString( 1 ))

assert(comment == actualRow.getString( 2 ))

assert(blank == actualRow.getString( 3 ))

}

test( " parse unescaped quotes with maxCharsPerColumn " ) {

val rows = spark.read

.format( " csv " )

.option( " maxCharsPerColumn " , " 4 " )

.load(testFile(unescapedQuotesFile))

val expectedRows = Seq ( Row ( " \" a \" b " , " ccc " , " ddd " ), Row ( " ab " , " cc \" c " , " ddd \" " ))

checkAnswer(rows, expectedRows)

}

test( " bad encoding name " ) {

val exception = intercept[ UnsupportedCharsetException ] {

spark

.read

.format( " csv " )

.option( " charset " , " 1-9588-osi " )

.load(testFile(carsFile8859))

}

assert(exception.getMessage.contains( " 1-9588-osi " ))

}

test( " test different encoding " ) {

withView( " carsTable " ) {

// scalastyle:off

spark.sql(

s """

|CREATE TEMPORARY VIEW carsTable USING csv

|OPTIONS (path " ${testFile(carsFile8859)} ", header "true",

|charset "iso-8859-1", delimiter "þ")

""" .stripMargin.replaceAll( " \n " , " " ))

// scalastyle:on

verifyCars(spark.table( " carsTable " ), withHeader = true )

}

test( " crlf line separators in multiline mode " ) {

val cars = spark

.read

.format( " csv " )

.option( " multiLine " , " true " )

.option( " header " , " true " )

.load(testFile(carsCrlf))

verifyCars(cars, withHeader = true )

}

test( " test aliases sep and encoding for delimiter and charset " ) {

// scalastyle:off

val cars = spark

.read

.format( " csv " )

.option( " header " , " true " )

.option( " encoding " , " iso-8859-1 " )

.option( " sep " , " þ " )

.load(testFile(carsFile8859))

// scalastyle:on

verifyCars(cars, withHeader = true )

}

test( " DDL test with tab separated file " ) {

withView( " carsTable " ) {

spark.sql(

s """

|CREATE TEMPORARY VIEW carsTable USING csv

|OPTIONS (path " ${testFile(carsTsvFile)} ", header "true", delimiter "\t")

""" .stripMargin.replaceAll( " \n " , " " ))

verifyCars(spark.table( " carsTable " ), numFields = 6 , withHeader = true , checkHeader = false )

}

test( " DDL test parsing decimal type " ) {

withView( " carsTable " ) {

spark.sql(

s """

|CREATE TEMPORARY VIEW carsTable

|(yearMade double, makeName string, modelName string, priceTag decimal,

| comments string, grp string)

|USING csv

|OPTIONS (path " ${testFile(carsTsvFile)} ", header "true", delimiter "\t")

""" .stripMargin.replaceAll( " \n " , " " ))

assert(

spark.sql( " SELECT makeName FROM carsTable where priceTag > 60000 " ).collect().length === 1 )

}

test( " test for DROPMALFORMED parsing mode " ) {

withSQLConf( SQLConf . CSV_PARSER_COLUMN_PRUNING .key - > " false " ) {

Seq ( false , true ).foreach { multiLine =>

val cars = spark.read

.format( " csv " )

.option( " multiLine " , multiLine)

.options( Map ( " header " - > " true " , " mode " - > " dropmalformed " ))

.load(testFile(carsFile))

assert(cars.select( " year " ).collect().length === 2 )

}

test( " when mode is null, will fall back to PermissiveMode mode " ) {

val cars = spark.read

.format( " csv " )

.options( Map ( " header " - > " true " , " mode " - > null ))

.load(testFile(carsFile))

assert(cars.collect().length == 3 )

assert(cars.select( " make " ).collect() sameElements

Array ( Row ( " Tesla " ), Row ( " Ford " ), Row ( " Chevy " )))

}

test( " test for blank column names on read and select columns " ) {

val cars = spark.read

.format( " csv " )

.options( Map ( " header " - > " true " , " inferSchema " - > " true " ))

.load(testFile(carsBlankColName))

assert(cars.select( " customer " ).collect().length == 2 )

assert(cars.select( " _c0 " ).collect().length == 2 )

assert(cars.select( " _c1 " ).collect().length == 2 )

}

test( " test for FAILFAST parsing mode " ) {

Seq ( false , true ).foreach { multiLine =>

val e1 = intercept[ SparkException ] {

spark.read

.format( " csv " )

.option( " multiLine " , multiLine)

.options( Map ( " header " - > " true " , " mode " - > " failfast " ))

.load(testFile(carsFile)).collect()

}

checkErrorMatchPVals(

exception = e1,

errorClass = " FAILED_READ_FILE.NO_HINT " ,

parameters = Map ( " path " - > s " .* $carsFile .* " ))

val e2 = e1.getCause. asInstanceOf [ SparkException ]

assert(e2.getErrorClass == " MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION " )

checkError(

exception = e2.getCause. asInstanceOf [ SparkRuntimeException ],

errorClass = " MALFORMED_CSV_RECORD " ,

parameters = Map ( " badRecord " - > " 2015,Chevy,Volt " )

)

}

test( " test for tokens more than the fields in the schema " ) {

val cars = spark

.read

.format( " csv " )

.option( " header " , " false " )

.option( " comment " , " ~ " )

.load(testFile(carsMalformedFile))

verifyCars(cars, withHeader = false , checkTypes = false )

}

test( " test with null quote character " ) {

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.option( " quote " , " " )

.load(testFile(carsUnbalancedQuotesFile))

verifyCars(cars, withHeader = true , checkValues = false )

test( " test with empty file and known schema " ) {

val result = spark.read

.format( " csv " )

.schema( StructType ( List ( StructField ( " column " , StringType , false ))))

.load(testFile(emptyFile))

assert(result.collect().length === 0 )

assert(result.schema.fieldNames.length === 1 )

}

test( " DDL test with empty file " ) {

withView( " carsTable " ) {

spark.sql(

s """

|CREATE TEMPORARY VIEW carsTable

|(yearMade double, makeName string, modelName string, comments string, grp string)

|USING csv

|OPTIONS (path " ${testFile(emptyFile)} ", header "false")

""" .stripMargin.replaceAll( " \n " , " " ))

assert(spark.sql( " SELECT count(*) FROM carsTable " ).collect().head( 0 ) === 0 )

}

test( " DDL test with schema " ) {

withView( " carsTable " ) {

spark.sql(

s """

|CREATE TEMPORARY VIEW carsTable

|(yearMade double, makeName string, modelName string, comments string, blank string)

|USING csv

|OPTIONS (path " ${testFile(carsFile)} ", header "true")

""" .stripMargin.replaceAll( " \n " , " " ))

val cars = spark.table( " carsTable " )

verifyCars(cars, withHeader = true , checkHeader = false , checkValues = false )

assert(

cars.schema.fieldNames === Array ( " yearMade " , " makeName " , " modelName " , " comments " , " blank " ))

}

test( " save csv " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.load(testFile(carsFile))

cars.coalesce( 1 ).write

.option( " header " , " true " )

.csv(csvDir)

val carsCopy = spark.read

.format( " csv " )

.option( " header " , " true " )

.load(csvDir)

verifyCars(carsCopy, withHeader = true )

}

test( " save csv with quote " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.load(testFile(carsFile))

cars.coalesce( 1 ).write

.format( " csv " )

.option( " header " , " true " )

.option( " quote " , " \" " )

.save(csvDir)

val carsCopy = spark.read

.format( " csv " )

.option( " header " , " true " )

.option( " quote " , " \" " )

.load(csvDir)

verifyCars(carsCopy, withHeader = true )

}

test( " save csv with quoteAll enabled " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val data = Seq (( " test \" quote \" " , 123 , " it \" works \" ! " , " \" very \" well " ))

val df = spark.createDataFrame(data)

// escapeQuotes should be true by default

df.coalesce( 1 ).write

.format( " csv " )

.option( " quote " , " \" " )

.option( " escape " , " \" " )

.option( " quoteAll " , " true " )

.save(csvDir)

val results = spark.read

.format( " text " )

.load(csvDir)

.collect()

val expected = " \" test \"\" quote \"\"\" , \" 123 \" , \" it \"\" works \"\" ! \" , \"\"\" very \"\" well \" "

assert(results.toSeq.map(_.toSeq) === Seq ( Seq (expected)))

}

test( " save csv with quote escaping enabled " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val data = Seq (( " test \" quote \" " , 123 , " it \" works \" ! " , " \" very \" well " ))

val df = spark.createDataFrame(data)

// escapeQuotes should be true by default

df.coalesce( 1 ).write

.format( " csv " )

.option( " quote " , " \" " )

.option( " escape " , " \" " )

.save(csvDir)

val results = spark.read

.format( " text " )

.load(csvDir)

.collect()

val expected = " \" test \"\" quote \"\"\" ,123, \" it \"\" works \"\" ! \" , \"\"\" very \"\" well \" "

assert(results.toSeq.map(_.toSeq) === Seq ( Seq (expected)))

}

test( " save csv with quote escaping disabled " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val data = Seq (( " test \" quote \" " , 123 , " it \" works \" ! " , " \" very \" well " ))

val df = spark.createDataFrame(data)

// escapeQuotes should be true by default

df.coalesce( 1 ).write

.format( " csv " )

.option( " quote " , " \" " )

.option( " escapeQuotes " , " false " )

.option( " escape " , " \" " )

.save(csvDir)

val results = spark.read

.format( " text " )

.load(csvDir)

.collect()

val expected = " test \" quote \" ,123,it \" works \" !, \"\"\" very \"\" well \" "

assert(results.toSeq.map(_.toSeq) === Seq ( Seq (expected)))

}

test( " save csv with quote escaping, using charToEscapeQuoteEscaping option " ) {

withTempPath { path =>

// original text

val df1 = Seq (

""" You are "beautiful" """ ,

""" Yes, \"in the inside"\ """

).toDF()

// text written in CSV with following options:

// quote character: "

// escape character: \

// character to escape quote escaping: #

val df2 = Seq (

""" "You are \"beautiful\"" """ ,

""" "Yes, # \\ "in the inside\"#\" """

).toDF()

df2.coalesce( 1 ).write.text(path.getAbsolutePath)

val df3 = spark.read

.format( " csv " )

.option( " quote " , " \" " )

.option( " escape " , " \\ " )

.option( " charToEscapeQuoteEscaping " , " # " )

.load(path.getAbsolutePath)

checkAnswer(df1, df3)

}

test( " SPARK-19018: Save csv with custom charset " ) {

// scalastyle:off nonascii

val content = " µß áâä ÁÂÄ "

// scalastyle:on nonascii

Seq ( " iso-8859-1 " , " utf-8 " , " utf-16 " , " utf-32 " , " windows-1250 " ).foreach { encoding =>

withTempPath { path =>

val csvDir = new File (path, " csv " )

Seq (content).toDF().write

.option( " encoding " , encoding)

.csv(csvDir.getCanonicalPath)

csvDir.listFiles().filter(_.getName.endsWith( " csv " )).foreach({ csvFile =>

val readback = Files .readAllBytes(csvFile.toPath)

val expected = (content + Properties .lineSeparator).getBytes( Charset .forName(encoding))

assert(readback === expected)

})

}

test( " SPARK-19018: error handling for unsupported charsets " ) {

val exception = intercept[ SparkException ] {

withTempPath { path =>

val csvDir = new File (path, " csv " ).getCanonicalPath

Seq ( " a,A,c,A,b,B " ).toDF().write

.option( " encoding " , " 1-9588-osi " )

.csv(csvDir)

}

assert(exception.getCause.getMessage.contains( " 1-9588-osi " ))

}

test( " commented lines in CSV data " ) {

Seq ( " false " , " true " ).foreach { multiLine =>

val results = spark.read

.format( " csv " )

.options( Map ( " comment " - > " ~ " , " header " - > " false " , " multiLine " - > multiLine))

.load(testFile(commentsFile))

.collect()

val expected =

Seq ( Seq ( " 1 " , " 2 " , " 3 " , " 4 " , " 5.01 " , " 2015-08-20 15:57:00 " ),

Seq ( " 6 " , " 7 " , " 8 " , " 9 " , " 0 " , " 2015-08-21 16:58:01 " ),

Seq ( " 1 " , " 2 " , " 3 " , " 4 " , " 5 " , " 2015-08-23 18:00:42 " ))

assert(results.toSeq.map(_.toSeq) === expected)

}

test( " inferring schema with commented lines in CSV data " ) {

val results = spark.read

.format( " csv " )

.options( Map ( " comment " - > " ~ " , " header " - > " false " , " inferSchema " - > " true " ))

.option( " timestampFormat " , " yyyy-MM-dd HH:mm:ss " )

.load(testFile(commentsFile))

.collect()

val expected =

Seq ( Seq ( 1 , 2 , 3 , 4 , 5.01D , Timestamp .valueOf( " 2015-08-20 15:57:00 " )),

Seq ( 6 , 7 , 8 , 9 , 0 , Timestamp .valueOf( " 2015-08-21 16:58:01 " )),

Seq ( 1 , 2 , 3 , 4 , 5 , Timestamp .valueOf( " 2015-08-23 18:00:42 " )))

assert(results.toSeq.map(_.toSeq) === expected)

}

test( " inferring timestamp types via custom date format " ) {

val options = Map (

" header " - > " true " ,

" inferSchema " - > " true " ,

" timestampFormat " - > " dd/MM/yyyy HH:mm " )

val results = spark.read

.format( " csv " )

.options(options)

.load(testFile(datesFile))

.select( " date " )

.collect()

val dateFormat = new SimpleDateFormat ( " dd/MM/yyyy HH:mm " , Locale . US )

val expected =

Seq ( Seq ( new Timestamp (dateFormat.parse( " 26/08/2015 18:00 " ).getTime)),

Seq ( new Timestamp (dateFormat.parse( " 27/10/2014 18:30 " ).getTime)),

Seq ( new Timestamp (dateFormat.parse( " 28/01/2016 20:00 " ).getTime)))

assert(results.toSeq.map(_.toSeq) === expected)

}

test( " load date types via custom date format " ) {

val customSchema = new StructType ( Array ( StructField ( " date " , DateType , true )))

val options = Map (

" header " - > " true " ,

" inferSchema " - > " false " ,

" dateFormat " - > " dd/MM/yyyy HH:mm " )

val results = spark.read

.format( " csv " )

.options(options)

.option( " timeZone " , " UTC " )

.schema(customSchema)

.load(testFile(datesFile))

.select( " date " )

.collect()

val dateFormat = new SimpleDateFormat ( " dd/MM/yyyy hh:mm " , Locale . US )

val expected = Seq (

new Date (dateFormat.parse( " 26/08/2015 18:00 " ).getTime),

new Date (dateFormat.parse( " 27/10/2014 18:30 " ).getTime),

new Date (dateFormat.parse( " 28/01/2016 20:00 " ).getTime))

val dates = results.toSeq.map(_.toSeq.head)

expected.zip(dates).foreach {

case (expectedDate, date) =>

// As it truncates the hours, minutes and etc., we only check

// if the dates (days, months and years) are the same via `toString()`.

assert(expectedDate.toString === date.toString)

}

test( " setting comment to null disables comment support " ) {

val results = spark.read

.format( " csv " )

.options( Map ( " comment " - > " " , " header " - > " false " ))

.load(testFile(disableCommentsFile))

.collect()

val expected =

Seq (

Seq ( " #1 " , " 2 " , " 3 " ),

Seq ( " 4 " , " 5 " , " 6 " ))

assert(results.toSeq.map(_.toSeq) === expected)

}

test( " nullable fields with user defined null value of \" null \" " ) {

// year,make,model,comment,blank

val dataSchema = StructType ( List (

StructField ( " year " , IntegerType , nullable = true ),

StructField ( " make " , StringType , nullable = false ),

StructField ( " model " , StringType , nullable = false ),

StructField ( " comment " , StringType , nullable = true ),

StructField ( " blank " , StringType , nullable = true )))

val cars = spark.read

.format( " csv " )

.schema(dataSchema)

.options( Map ( " header " - > " true " , " nullValue " - > " null " ))

.load(testFile(carsNullFile))

verifyCars(cars, withHeader = true , checkValues = false )

val results = cars.collect()

assert(results( 0 ).toSeq === Array ( 2012 , " Tesla " , " S " , null , null ))

assert(results( 2 ).toSeq === Array ( null , " Chevy " , " Volt " , null , null ))

}

test( " empty fields with user defined empty values " ) {

// year,make,model,comment,blank

val dataSchema = StructType ( List (

StructField ( " year " , IntegerType , nullable = true ),

StructField ( " make " , StringType , nullable = false ),

StructField ( " model " , StringType , nullable = false ),

StructField ( " comment " , StringType , nullable = true ),

StructField ( " blank " , StringType , nullable = true )))

val cars = spark.read

.format( " csv " )

.schema(dataSchema)

.option( " header " , " true " )

.option( " emptyValue " , " empty " )

.load(testFile(carsEmptyValueFile))

verifyCars(cars, withHeader = true , checkValues = false )

val results = cars.collect()

assert(results( 0 ).toSeq === Array ( 2012 , " Tesla " , " S " , " empty " , " empty " ))

assert(results( 1 ).toSeq ===

Array ( 1997 , " Ford " , " E350 " , " Go get one now they are going fast " , null ))

assert(results( 2 ).toSeq === Array ( 2015 , " Chevy " , " Volt " , null , " empty " ))

}

test( " save csv with empty fields with user defined empty values " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

// year,make,model,comment,blank

val dataSchema = StructType ( List (

StructField ( " year " , IntegerType , nullable = true ),

StructField ( " make " , StringType , nullable = false ),

StructField ( " model " , StringType , nullable = false ),

StructField ( " comment " , StringType , nullable = true ),

StructField ( " blank " , StringType , nullable = true )))

val cars = spark.read

.format( " csv " )

.schema(dataSchema)

.option( " header " , " true " )

.option( " nullValue " , " NULL " )

.load(testFile(carsEmptyValueFile))

cars.coalesce( 1 ).write

.format( " csv " )

.option( " header " , " true " )

.option( " emptyValue " , " empty " )

.option( " nullValue " , null )

.save(csvDir)

val carsCopy = spark.read

.format( " csv " )

.schema(dataSchema)

.option( " header " , " true " )

.load(csvDir)

verifyCars(carsCopy, withHeader = true , checkValues = false )

val results = carsCopy.collect()

assert(results( 0 ).toSeq === Array ( 2012 , " Tesla " , " S " , " empty " , " empty " ))

assert(results( 1 ).toSeq ===

Array ( 1997 , " Ford " , " E350 " , " Go get one now they are going fast " , null ))

assert(results( 2 ).toSeq === Array ( 2015 , " Chevy " , " Volt " , null , " empty " ))

}

test( " SPARK-37575: null values should be saved as nothing rather than " +

" quoted empty Strings \"\" with default settings " ) {

Seq ( " true " , " false " ).foreach { confVal =>

withSQLConf( SQLConf . LEGACY_NULL_VALUE_WRITTEN_AS_QUOTED_EMPTY_STRING_CSV .key - > confVal) {

withTempPath { path =>

Seq (( " Tesla " , null : String , " " ))

.toDF( " make " , " comment " , " blank " )

.write

.csv(path.getCanonicalPath)

if (confVal == " false " ) {

checkAnswer(spark.read.text(path.getCanonicalPath), Row ( " Tesla,, \"\" " ))

} else {

checkAnswer(spark.read.text(path.getCanonicalPath), Row ( " Tesla, \"\" , \"\" " ))

}

test( " save csv with compression codec option " ) {

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.load(testFile(carsFile))

cars.coalesce( 1 ).write

.format( " csv " )

.option( " header " , " true " )

.option( " compression " , " gZiP " )

.save(csvDir)

val compressedFiles = new File (csvDir).listFiles()

assert(compressedFiles.exists(_.getName.endsWith( " .csv.gz " )))

val carsCopy = spark.read

.format( " csv " )

.option( " header " , " true " )

.load(csvDir)

verifyCars(carsCopy, withHeader = true )

}

test( " SPARK-13543 Write the output as uncompressed via option() " ) {

val extraOptions = Map (

" mapreduce.output.fileoutputformat.compress " - > " true " ,

" mapreduce.output.fileoutputformat.compress.type " - > CompressionType . BLOCK .toString,

" mapreduce.map.output.compress " - > " true " ,

" mapreduce.map.output.compress.codec " - > classOf [ GzipCodec ].getName

)

withTempDir { dir =>

val csvDir = new File (dir, " csv " ).getCanonicalPath

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.options(extraOptions)

.load(testFile(carsFile))

cars.coalesce( 1 ).write

.format( " csv " )

.option( " header " , " true " )

.option( " compression " , HadoopCompressionCodec . NONE .lowerCaseName())

.options(extraOptions)

.save(csvDir)

val compressedFiles = new File (csvDir).listFiles()

assert(compressedFiles.exists( ! _.getName.endsWith( " .csv.gz " )))

val carsCopy = spark.read

.format( " csv " )

.option( " header " , " true " )

.options(extraOptions)

.load(csvDir)

verifyCars(carsCopy, withHeader = true )

}

test( " Schema inference correctly identifies the datatype when data is sparse. " ) {

val df = spark.read

.format( " csv " )

.option( " header " , " true " )

.option( " inferSchema " , " true " )

.load(testFile(simpleSparseFile))

assert(

df.schema.fields.map(field => field.dataType).sameElements(

Array ( IntegerType , IntegerType , IntegerType , IntegerType )))

}

test( " old csv data source name works " ) {

val cars = spark

.read

.format( " com.databricks.spark.csv " )

.option( " header " , " false " )

.load(testFile(carsFile))

verifyCars(cars, withHeader = false , checkTypes = false )

}

test( " nulls, NaNs and Infinity values can be parsed " ) {

val numbers = spark

.read

.format( " csv " )

.schema( StructType ( List (

StructField ( " int " , IntegerType , true ),

StructField ( " long " , LongType , true ),

StructField ( " float " , FloatType , true ),

StructField ( " double " , DoubleType , true )

)))

.options( Map (

" header " - > " true " ,

" mode " - > " DROPMALFORMED " ,

" nullValue " - > " -- " ,

" nanValue " - > " NAN " ,

" negativeInf " - > " -INF " ,

" positiveInf " - > " INF " ))

.load(testFile(numbersFile))

assert(numbers.count() == 8 )

}

test( " SPARK-15585 turn off quotations " ) {

val cars = spark.read

.format( " csv " )

.option( " header " , " true " )

.option( " quote " , " " )

.load(testFile(carsUnbalancedQuotesFile))

verifyCars(cars, withHeader = true , checkValues = false )

}

test( " Write timestamps correctly in ISO8601 format by default " ) {

withTempDir { dir =>

val iso8601timestampsPath = s " ${dir.getCanonicalPath} /iso8601timestamps.csv "

val timestamps = spark.read

.format( " csv " )

.option( " inferSchema " , " true " )

.option( " header " , " true " )

.option( " timestampFormat " , " dd/MM/yyyy HH:mm " )

.load(testFile(datesFile))

timestamps.write

.format( " csv " )

.option( " header " , " true " )

.save(iso8601timestampsPath)

// This will load back the timestamps as string.

val stringSchema = StructType ( StructField ( " date " , StringType , true ) :: Nil )

val iso8601Timestamps = spark.read

.format( " csv " )

.schema(stringSchema)

.option( " header " , " true " )

.load(iso8601timestampsPath)

val iso8501 = FastDateFormat .getInstance( " yyyy-MM-dd'T'HH:mm:ss.SSSXXX " , Locale . US )

val expectedTimestamps = timestamps.collect().map { r =>

// This should be ISO8601 formatted string.

Row (iso8501.format(r.toSeq.head))

}

checkAnswer(iso8601Timestamps, expectedTimestamps)

}

test( " Write dates correctly in ISO8601 format by default " ) {

withSQLConf( SQLConf . SESSION_LOCAL_TIMEZONE .key - > " UTC " ) {

withTempDir { dir =>

val customSchema = new StructType ( Array ( StructField ( " date " , DateType , true )))

val iso8601datesPath = s " ${dir.getCanonicalPath} /iso8601dates.csv "

View remainder of file in raw view

CSVSuite.scala

Latest commit

History

CSVSuite.scala

File metadata and controls