From 547f8228e070e508770c12195abaa84b191aa6c1 Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Thu, 11 Feb 2016 17:55:28 -0500 Subject: [PATCH 1/6] Fixed-width parsing spark integration. Based on and heavily reliant on databricks-spark-csv (a slightly modified version for now) Initial commit with basic tests. TODO: more tests --- .gitignore | 1 + build.sbt | 12 ++ project/assembly.sbt | 1 + project/build.properties | 1 + project/plugins.sbt | 1 + .../spark/fixedwidth/FixedwidthRelation.scala | 53 +++++++ .../spark/fixedwidth/package.scala | 36 +++++ .../spark/fixedwidth/readers/readers.scala | 142 ++++++++++++++++++ src/test/resources/fruit__fixedwidth.txt | 7 + .../resources/fruit_overflow_fixedwidth.txt | 7 + .../resources/fruit_underflow_fixedwidth.txt | 7 + .../resources/fruit_w_headers_fixedwidth.txt | 8 + .../spark/fixedwidth/FixedwidthSuite.scala | 71 +++++++++ 13 files changed, 347 insertions(+) create mode 100644 build.sbt create mode 100644 project/assembly.sbt create mode 100644 project/build.properties create mode 100644 project/plugins.sbt create mode 100644 src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala create mode 100644 src/main/scala/com/quartethealth/spark/fixedwidth/package.scala create mode 100644 src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala create mode 100644 src/test/resources/fruit__fixedwidth.txt create mode 100644 src/test/resources/fruit_overflow_fixedwidth.txt create mode 100644 src/test/resources/fruit_underflow_fixedwidth.txt create mode 100644 src/test/resources/fruit_w_headers_fixedwidth.txt create mode 100644 src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala diff --git a/.gitignore b/.gitignore index c58d83b..52cb35b 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ .cache .history .lib/ +.idea/ dist/* target/ lib_managed/ diff --git a/build.sbt b/build.sbt new file mode 100644 index 0000000..6a200bd --- /dev/null +++ b/build.sbt @@ -0,0 +1,12 @@ +name := "spark-fixedwidth" + +version := "1.0" + +scalaVersion := "2.11.7" + +libraryDependencies ++= Seq( + "com.univocity" % "univocity-parsers" % "1.5.1", + "org.apache.spark" %% "spark-sql" % "1.6.0" % "provided", + "org.apache.spark" %% "spark-core" % "1.6.0" % "provided", + "org.scalatest" %% "scalatest" % "2.2.1" % "test" +) \ No newline at end of file diff --git a/project/assembly.sbt b/project/assembly.sbt new file mode 100644 index 0000000..74adde3 --- /dev/null +++ b/project/assembly.sbt @@ -0,0 +1 @@ +addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.13.0") diff --git a/project/build.properties b/project/build.properties new file mode 100644 index 0000000..d638b4f --- /dev/null +++ b/project/build.properties @@ -0,0 +1 @@ +sbt.version = 0.13.8 \ No newline at end of file diff --git a/project/plugins.sbt b/project/plugins.sbt new file mode 100644 index 0000000..14a6ca1 --- /dev/null +++ b/project/plugins.sbt @@ -0,0 +1 @@ +logLevel := Level.Warn \ No newline at end of file diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala new file mode 100644 index 0000000..b69d83e --- /dev/null +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala @@ -0,0 +1,53 @@ +package com.quartethealth.spark.fixedwidth + +import com.databricks.spark.csv.readers.{BulkReader, LineReader} +import com.quartethealth.spark.fixedwidth.readers.{LineFixedwidthReader, BulkFixedwidthReader} +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.SQLContext + +import com.databricks.spark.csv.CsvRelation + +class FixedwidthRelation protected[spark] ( + baseRDD: () => RDD[String], + fixedWidths: Array[Int], + location: Option[String], + useHeader: Boolean, + parseMode: String, + comment: Character, + ignoreLeadingWhiteSpace: Boolean, + ignoreTrailingWhiteSpace: Boolean, + treatEmptyValuesAsNulls: Boolean, + userSchema: StructType, + inferSchema: Boolean, + codec: String = null, + nullValue: String = "")(@transient override val sqlContext: SQLContext) + extends CsvRelation( + baseRDD, + location, + useHeader, + delimiter = '\0', + quote = null, + escape = null, + comment = comment, + parseMode = parseMode, + parserLib = "UNIVOCITY", + ignoreLeadingWhiteSpace = ignoreLeadingWhiteSpace, + ignoreTrailingWhiteSpace = ignoreTrailingWhiteSpace, + treatEmptyValuesAsNulls = treatEmptyValuesAsNulls, + userSchema = userSchema, + inferCsvSchema = true, + codec = codec)(sqlContext) { + + protected override def getLineReader(): LineReader = { + val commentChar: Char = if (comment == null) '\0' else comment + + new LineFixedwidthReader(fixedWidths, commentMarker = commentChar) + } + + protected override def getBulkReader(header: Seq[String], iter: Iterator[String], split: Int): BulkReader = { + val commentChar: Char = if (comment == null) '\0' else comment + new BulkFixedwidthReader(iter, split, fixedWidths, + headers = header, commentMarker = commentChar) + } +} diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala new file mode 100644 index 0000000..ec2c69a --- /dev/null +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala @@ -0,0 +1,36 @@ +package com.quartethealth.spark + + +import com.databricks.spark.csv.util.TextFile +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.{DataFrame, SQLContext} + +package object fixedwidth { + implicit class CsvContext(sqlContext: SQLContext) extends Serializable { + def flatFile( + filePath: String, + fixedWidths: Array[Int], + schema: StructType = null, + useHeader: Boolean = true, + mode: String = "PERMISSIVE", + comment: Character = null, + ignoreLeadingWhiteSpace: Boolean = false, + ignoreTrailingWhiteSpace: Boolean = false, + charset: String = TextFile.DEFAULT_CHARSET.name(), + inferSchema: Boolean = false): DataFrame = { + val fixedwidthRelation = new FixedwidthRelation( + () => TextFile.withCharset(sqlContext.sparkContext, filePath, charset), + location = Some(filePath), + useHeader = useHeader, + comment = comment, + parseMode = mode, + fixedWidths = fixedWidths, + ignoreLeadingWhiteSpace = ignoreLeadingWhiteSpace, + ignoreTrailingWhiteSpace = ignoreTrailingWhiteSpace, + userSchema = schema, + inferSchema = inferSchema, + treatEmptyValuesAsNulls = false)(sqlContext) + sqlContext.baseRelationToDataFrame(fixedwidthRelation) + } + } +} diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala new file mode 100644 index 0000000..f6843a8 --- /dev/null +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala @@ -0,0 +1,142 @@ +package com.quartethealth.spark.fixedwidth.readers + +import com.databricks.spark.csv.readers.{BulkReader, LineReader} +import com.univocity.parsers.fixed.{FixedWidthParserSettings, FixedWidthFieldLengths, FixedWidthParser} + +/** + * Read and parse Fixed-width-like input + * + * @param fixedWidths the fixed widths of the fields + * @param lineSep the delimiter used to separate lines + * @param commentMarker Ignore lines starting with this char + * @param ignoreLeadingSpace ignore white space before a field + * @param ignoreTrailingSpace ignore white space after a field + * @param headers headers for the columns + * @param inputBufSize size of buffer to use for parsing input, tune for performance + * @param maxCols maximum number of columns allowed, for safety against bad inputs + */ +private[readers] abstract class FixedwidthReader( + fixedWidths: Array[Int], + lineSep: String = "\n", + commentMarker: Char = '#', + ignoreLeadingSpace: Boolean = true, + ignoreTrailingSpace: Boolean = true, + headers: Seq[String], + inputBufSize: Int = 128, + maxCols: Int = 20480) { + protected lazy val parser: FixedWidthParser = { + val settings = new FixedWidthParserSettings(new FixedWidthFieldLengths(fixedWidths: _*)) + val format = settings.getFormat + format.setLineSeparator(lineSep) + format.setComment(commentMarker) + settings.setIgnoreLeadingWhitespaces(ignoreLeadingSpace) + settings.setIgnoreTrailingWhitespaces(ignoreTrailingSpace) + settings.setReadInputOnSeparateThread(false) + settings.setInputBufferSize(inputBufSize) + settings.setMaxColumns(maxCols) + settings.setNullValue("") + settings.setMaxCharsPerColumn(100000) + if (headers != null) settings.setHeaders(headers: _*) + // TODO: configurable? + settings.setSkipTrailingCharsUntilNewline(true) + settings.setRecordEndsOnNewline(true) + + new FixedWidthParser(settings) + } +} + +/** + * Read and parse a single line of Fixed-width-like input. Inefficient for bulk data. + * @param fixedWidths the fixed widths of the fields + * @param lineSep the delimiter used to separate lines + * @param commentMarker Ignore lines starting with this char + * @param ignoreLeadingSpace ignore white space before a field + * @param ignoreTrailingSpace ignore white space after a field + * @param inputBufSize size of buffer to use for parsing input, tune for performance + * @param maxCols maximum number of columns allowed, for safety against bad inputs + */ +private[fixedwidth] class LineFixedwidthReader( + fixedWidths: Array[Int], + lineSep: String = "\n", + commentMarker: Char = '#', + ignoreLeadingSpace: Boolean = true, + ignoreTrailingSpace: Boolean = true, + inputBufSize: Int = 128, + maxCols: Int = 20480) + extends FixedwidthReader( + fixedWidths, + lineSep, + commentMarker, + ignoreLeadingSpace, + ignoreTrailingSpace, + null, + inputBufSize, + maxCols) + with LineReader { + /** + * parse a line + * @param line a String with no newline at the end + * @return array of strings where each string is a field in the CSV record + */ + def parseLine(line: String): Array[String] = { + parser.beginParsing(getReader(line)) + val parsed = parser.parseNext() + parser.stopParsing() + parsed + } +} + +/** + * Read and parse Fixed-width-like input + * + * @param fixedWidths the fixed widths of the fields + * @param lineSep the delimiter used to separate lines + * @param commentMarker Ignore lines starting with this char + * @param ignoreLeadingSpace ignore white space before a field + * @param ignoreTrailingSpace ignore white space after a field + * @param headers headers for the columns + * @param inputBufSize size of buffer to use for parsing input, tune for performance + * @param maxCols maximum number of columns allowed, for safety against bad inputs + */ +private[fixedwidth] class BulkFixedwidthReader( + iter: Iterator[String], + split: Int, // for debugging + fixedWidths: Array[Int], + lineSep: String = "\n", + commentMarker: Char = '#', + ignoreLeadingSpace: Boolean = true, + ignoreTrailingSpace: Boolean = true, + headers: Seq[String], + inputBufSize: Int = 128, + maxCols: Int = 20480) + extends FixedwidthReader( + fixedWidths, + lineSep, + commentMarker, + ignoreLeadingSpace, + ignoreTrailingSpace, + headers, + inputBufSize, + maxCols + ) with BulkReader { + + parser.beginParsing(getReader(iter)) + private var nextRecord = parser.parseNext() + + /** + * get the next parsed line. + * + * @return array of strings where each string is a field in the fixed-width record + */ + override def next(): Array[String] = { + val curRecord = nextRecord + if(curRecord != null) { + nextRecord = parser.parseNext() + } else { + throw new NoSuchElementException("next record is null") + } + curRecord + } + + override def hasNext: Boolean = nextRecord != null +} diff --git a/src/test/resources/fruit__fixedwidth.txt b/src/test/resources/fruit__fixedwidth.txt new file mode 100644 index 0000000..157823e --- /dev/null +++ b/src/test/resources/fruit__fixedwidth.txt @@ -0,0 +1,7 @@ +56 apple TRUE 0.56 +45 pear FALSE1.34 +34 raspberry TRUE 2.43 +34 plum TRUE 1.31 +53 cherry TRUE 1.4 +23 orange FALSE2.34 +56 persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/resources/fruit_overflow_fixedwidth.txt b/src/test/resources/fruit_overflow_fixedwidth.txt new file mode 100644 index 0000000..38b97fe --- /dev/null +++ b/src/test/resources/fruit_overflow_fixedwidth.txt @@ -0,0 +1,7 @@ +56 apple TRUE 0.56 +45 pear FALSE1.34 +34 raspberry TRUE 2.436565 +34 plum TRUE 1.31 +53 cherry TRUE 1.4 +23 orange FALSE2.3466 +56 persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/resources/fruit_underflow_fixedwidth.txt b/src/test/resources/fruit_underflow_fixedwidth.txt new file mode 100644 index 0000000..a41698e --- /dev/null +++ b/src/test/resources/fruit_underflow_fixedwidth.txt @@ -0,0 +1,7 @@ +56 apple TRUE 0.56 +45 pear FALSE1.34 +34 raspberry TRUE 2 +34 plum TRUE 1.31 +53 cherry TRUE 1 +23 orange FALSE2.34 +56 persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/resources/fruit_w_headers_fixedwidth.txt b/src/test/resources/fruit_w_headers_fixedwidth.txt new file mode 100644 index 0000000..dd7ecb2 --- /dev/null +++ b/src/test/resources/fruit_w_headers_fixedwidth.txt @@ -0,0 +1,8 @@ +AMTNAME SALE COST +56 apple TRUE 0.56 +45 pear FALSE1.34 +34 raspberry TRUE 2.43 +34 plum TRUE 1.31 +53 cherry TRUE 1.4 +23 orange FALSE2.34 +56 persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala new file mode 100644 index 0000000..88aa225 --- /dev/null +++ b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala @@ -0,0 +1,71 @@ +package com.quartethealth.spark.fixedwidth + + +import org.apache.spark.SparkContext +import org.apache.spark.sql.{DataFrame, SQLContext} +import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType}; +import org.scalatest.{BeforeAndAfterAll, FunSuite} + +class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { + def fruit_resource(name: String = ""): String = s"src/test/resources/fruit_${name}_fixedwidth.txt" + + val fruitWidths = Array(3, 10, 5, 4) + val fruitSize = 7 + val fruitFirstRow = Seq(56, "apple", "TRUE", 0.56) + + val fruitSchema = StructType(Seq( + StructField("val", IntegerType), + StructField("name", StringType), + StructField("avail", StringType), + StructField("cost", DoubleType) + )) + + private var sqlContext: SQLContext = _ + + override protected def beforeAll(): Unit = { + super.beforeAll() + sqlContext = new SQLContext(new SparkContext("local[2]", "FixedwidthSuite")) + } + + override protected def afterAll(): Unit = { + try { + sqlContext.sparkContext.stop() + } finally { + super.afterAll() + } + } + + private def sanityChecks(resultSet: DataFrame): Unit = { + resultSet.show() + assert(resultSet.collect().length === fruitSize) + + val head = resultSet.head() + assert(head.length === fruitWidths.length) + assert(head.toSeq === fruitFirstRow) + } + + test("Parse basic") { + val result = sqlContext.flatFile(fruit_resource(), fruitWidths, fruitSchema, + useHeader = false) + sanityChecks(result) + } + + test("Parse with headers, ignore") { + val result = sqlContext.flatFile(fruit_resource("w_headers"), fruitWidths, + fruitSchema, useHeader = true) + sanityChecks(result) + } + + test("Parse with overflow, ignore") { + val result = sqlContext.flatFile(fruit_resource("overflow"), fruitWidths, + fruitSchema, useHeader = false) + sanityChecks(result) + } + + test("Parse with underflow, ignore") { + val result = sqlContext.flatFile(fruit_resource("underflow"), fruitWidths, + fruitSchema, useHeader = false) + sanityChecks(result) + } + +} From 38f0aac2ef370786ebf70b58748536ce0899a179 Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Thu, 11 Feb 2016 18:36:51 -0500 Subject: [PATCH 2/6] Added docs. Some renaming --- README.md | 55 ++++++++++++++++++- .../spark/fixedwidth/package.scala | 2 +- .../spark/fixedwidth/FixedwidthSuite.scala | 10 ++-- 3 files changed, 60 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index c8ef600..355267c 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,55 @@ # spark-fixedwidth -Fixed-width data source for Spark SQL and DataFrames +Fixed-width data source for Spark SQL and DataFrames. Based on (and uses) [databricks-spark-csv](https://github.com/databricks/spark-csv) + +## Requirements +This library requires Spark 1.3+ and Scala 2.11+ + +## Building +Run `sbt assembly` from inside the root directory to generate a JAR + +## Running / Using + +### In the Spark Shell +`./bin/spark-shell --jars /spark-fixedwidth/target/scala-2.11/spark-fixedwidth-assembly-1.0.jar` + +### In another project +Add the JAR to your project lib and sbt will include it for you + +## Features +This package allows reading fixed-width files in local or distributed filesystem as [Spark DataFrames](https://spark.apache.org/docs/1.3.0/sql-programming-guide.html). +When reading files the API accepts several options: +* `path` (REQUIRED): location of files. Similar to Spark can accept standard Hadoop globbing expressions. +* `fixedWidths` (REQUIRED): Int array of the fixed widths of the source file(s) +* `schema`: in [spark SQL form](http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.types.StructType). Otherwise everything is assumed String (unless inferSchema is on) +* `useHeader`: when set to true the first line of files will be used to name columns and will not be included in data. All types will be assumed string. Default value is true. +* `charset`: defaults to 'UTF-8' but can be set to other valid charset names +* `inferSchema`: automatically infers column types. It requires one extra pass over the data and is false by default +* `comment`: skip lines beginning with this character. Default is `"#"`. Disable comments by setting this to `null`. +* `codec`: compression codec to use when saving to file. Should be the fully qualified name of a class implementing `org.apache.hadoop.io.compress.CompressionCodec` or one of case-insensitive shorten names (`bzip2`, `gzip`, `lz4`, and `snappy`). Defaults to no compression when a codec is not specified. +* `nullValue`: specificy a string that indicates a null value, any fields matching this string will be set as nulls in the DataFrame + +### Scala API +__Spark 1.4+:__ +```scala +import org.apache.spark.sql.{DataFrame, SQLContext} +import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} + +val fruitSchema = StructType(Seq( + StructField("val", IntegerType), + StructField("name", StringType), + StructField("avail", StringType), + StructField("cost", DoubleType) +)) + +val sqlContext = new SQLContext(sc) +val fruitWidths = Array(3, 10, 5, 4) +val fruit_resource = 'fruit_fixedwidths.txt' + +val result = sqlContext.fixedFile( + fruit_resource, + fruitWidths, + fruitSchema, + useHeader = false +) +result.show() // Prints top 20 rows in tabular format +``` \ No newline at end of file diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala index ec2c69a..b75af89 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala @@ -7,7 +7,7 @@ import org.apache.spark.sql.{DataFrame, SQLContext} package object fixedwidth { implicit class CsvContext(sqlContext: SQLContext) extends Serializable { - def flatFile( + def fixedFile( filePath: String, fixedWidths: Array[Int], schema: StructType = null, diff --git a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala index 88aa225..2125783 100644 --- a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala +++ b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala @@ -3,7 +3,7 @@ package com.quartethealth.spark.fixedwidth import org.apache.spark.SparkContext import org.apache.spark.sql.{DataFrame, SQLContext} -import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType}; +import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} import org.scalatest.{BeforeAndAfterAll, FunSuite} class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { @@ -45,25 +45,25 @@ class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { } test("Parse basic") { - val result = sqlContext.flatFile(fruit_resource(), fruitWidths, fruitSchema, + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, fruitSchema, useHeader = false) sanityChecks(result) } test("Parse with headers, ignore") { - val result = sqlContext.flatFile(fruit_resource("w_headers"), fruitWidths, + val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, fruitSchema, useHeader = true) sanityChecks(result) } test("Parse with overflow, ignore") { - val result = sqlContext.flatFile(fruit_resource("overflow"), fruitWidths, + val result = sqlContext.fixedFile(fruit_resource("overflow"), fruitWidths, fruitSchema, useHeader = false) sanityChecks(result) } test("Parse with underflow, ignore") { - val result = sqlContext.flatFile(fruit_resource("underflow"), fruitWidths, + val result = sqlContext.fixedFile(fruit_resource("underflow"), fruitWidths, fruitSchema, useHeader = false) sanityChecks(result) } From d3d3bc5edaa9b913ea64e293d5e3adf76e3c41be Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Fri, 12 Feb 2016 13:31:21 -0500 Subject: [PATCH 3/6] More tests and readme Also made the whitespace options actually useable (databricks spark-csv exposes but ignores these). --- README.md | 26 +++++++- .../spark/fixedwidth/FixedwidthRelation.scala | 11 +++- .../spark/fixedwidth/package.scala | 4 +- .../resources/fruit_comments_fixedwidth.txt | 11 ++++ .../resources/fruit_malformed_fixedwidth.txt | 7 +++ .../spark/fixedwidth/FixedwidthSuite.scala | 59 ++++++++++++++++--- 6 files changed, 104 insertions(+), 14 deletions(-) create mode 100644 src/test/resources/fruit_comments_fixedwidth.txt create mode 100644 src/test/resources/fruit_malformed_fixedwidth.txt diff --git a/README.md b/README.md index 355267c..97814a9 100644 --- a/README.md +++ b/README.md @@ -25,11 +25,19 @@ When reading files the API accepts several options: * `charset`: defaults to 'UTF-8' but can be set to other valid charset names * `inferSchema`: automatically infers column types. It requires one extra pass over the data and is false by default * `comment`: skip lines beginning with this character. Default is `"#"`. Disable comments by setting this to `null`. +* `mode`: determines the parsing mode. By default it is PERMISSIVE. Possible values are: + * `PERMISSIVE`: tries to parse all lines: nulls are inserted for missing tokens and extra tokens are ignored. + * `DROPMALFORMED`: drops lines which have fewer or more tokens than expected or tokens which do not match the schema + * `FAILFAST`: aborts with a RuntimeException if encounters any malformed line * `codec`: compression codec to use when saving to file. Should be the fully qualified name of a class implementing `org.apache.hadoop.io.compress.CompressionCodec` or one of case-insensitive shorten names (`bzip2`, `gzip`, `lz4`, and `snappy`). Defaults to no compression when a codec is not specified. -* `nullValue`: specificy a string that indicates a null value, any fields matching this string will be set as nulls in the DataFrame +* `nullValue`: specify a string that indicates a null value. Any fields matching this string will be set as nulls in the DataFrame +* `ignoreLeadingWhiteSpace`: Boolean, default true +* `ignoreTrailingWhiteSpace`: Boolean, default true ### Scala API __Spark 1.4+:__ + +See [sample fixed-width files](src/test/resources) ```scala import org.apache.spark.sql.{DataFrame, SQLContext} import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} @@ -52,4 +60,18 @@ val result = sqlContext.fixedFile( useHeader = false ) result.show() // Prints top 20 rows in tabular format -``` \ No newline at end of file + +// Example without schema, and showing extra options +val fruit_resource = 'fruit_w_headers_fixedwidths.txt' +val result = sqlContext.fixedFile( + fruit_resource, + fruitWidths, + useHeader = true, + inferSchema = true, + mode = "DROPMALFORMED", + comment = '/', + ignoreLeadingWhiteSpace: true, + ignoreTrailingWhiteSpace: false, +) +result.collect() // Returns an array that contains all of Rows in this DataFrame +``` diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala index b69d83e..8410f8a 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala @@ -42,12 +42,17 @@ class FixedwidthRelation protected[spark] ( protected override def getLineReader(): LineReader = { val commentChar: Char = if (comment == null) '\0' else comment - new LineFixedwidthReader(fixedWidths, commentMarker = commentChar) + new LineFixedwidthReader(fixedWidths, commentMarker = commentChar, + ignoreLeadingSpace = ignoreLeadingWhiteSpace, + ignoreTrailingSpace = ignoreTrailingWhiteSpace) } - protected override def getBulkReader(header: Seq[String], iter: Iterator[String], split: Int): BulkReader = { + protected override def getBulkReader(header: Seq[String], iter: Iterator[String], + split: Int): BulkReader = { val commentChar: Char = if (comment == null) '\0' else comment new BulkFixedwidthReader(iter, split, fixedWidths, - headers = header, commentMarker = commentChar) + headers = header, commentMarker = commentChar, + ignoreLeadingSpace = ignoreLeadingWhiteSpace, + ignoreTrailingSpace = ignoreTrailingWhiteSpace) } } diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala index b75af89..7f3e50a 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala @@ -14,8 +14,8 @@ package object fixedwidth { useHeader: Boolean = true, mode: String = "PERMISSIVE", comment: Character = null, - ignoreLeadingWhiteSpace: Boolean = false, - ignoreTrailingWhiteSpace: Boolean = false, + ignoreLeadingWhiteSpace: Boolean = true, + ignoreTrailingWhiteSpace: Boolean = true, charset: String = TextFile.DEFAULT_CHARSET.name(), inferSchema: Boolean = false): DataFrame = { val fixedwidthRelation = new FixedwidthRelation( diff --git a/src/test/resources/fruit_comments_fixedwidth.txt b/src/test/resources/fruit_comments_fixedwidth.txt new file mode 100644 index 0000000..18faa5b --- /dev/null +++ b/src/test/resources/fruit_comments_fixedwidth.txt @@ -0,0 +1,11 @@ +// Fruit for sale +// 2014-10-02 +AMTNAME AVAILCOST +56 apple TRUE 0.56 +45 pear FALSE1.34 +34 raspberry TRUE 2.43 +34 plum TRUE 1.31 +53 cherry TRUE 1.4 +23 orange FALSE2.34 +56 persimmon FALSE23.2 +// No more fruit to show \ No newline at end of file diff --git a/src/test/resources/fruit_malformed_fixedwidth.txt b/src/test/resources/fruit_malformed_fixedwidth.txt new file mode 100644 index 0000000..8418226 --- /dev/null +++ b/src/test/resources/fruit_malformed_fixedwidth.txt @@ -0,0 +1,7 @@ +23 apple TRUE 1.44 +45 cherry FALSE1.33 +34 raspberry TRUE BLR +34 +ewfergerfgrefergregreg +hi pear +56 persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala index 2125783..ba4cdf3 100644 --- a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala +++ b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala @@ -7,20 +7,20 @@ import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerT import org.scalatest.{BeforeAndAfterAll, FunSuite} class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { - def fruit_resource(name: String = ""): String = s"src/test/resources/fruit_${name}_fixedwidth.txt" + protected def fruit_resource(name: String = ""): String = s"src/test/resources/fruit_${name}_fixedwidth.txt" - val fruitWidths = Array(3, 10, 5, 4) - val fruitSize = 7 - val fruitFirstRow = Seq(56, "apple", "TRUE", 0.56) + protected val fruitWidths = Array(3, 10, 5, 4) + protected val fruitSize = 7 + protected val fruitFirstRow = Seq(56, "apple", "TRUE", 0.56) - val fruitSchema = StructType(Seq( + protected val fruitSchema = StructType(Seq( StructField("val", IntegerType), StructField("name", StringType), StructField("avail", StringType), StructField("cost", DoubleType) )) - private var sqlContext: SQLContext = _ + protected var sqlContext: SQLContext = _ override protected def beforeAll(): Unit = { super.beforeAll() @@ -35,7 +35,7 @@ class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { } } - private def sanityChecks(resultSet: DataFrame): Unit = { + protected def sanityChecks(resultSet: DataFrame): Unit = { resultSet.show() assert(resultSet.collect().length === fruitSize) @@ -68,4 +68,49 @@ class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { sanityChecks(result) } + test("Parse basic without schema, uninferred") { + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, + useHeader = false, inferSchema = false) + sanityChecks(result) + } + + test("Parse basic without schema, inferred") { + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, + useHeader = false, inferSchema = true) + sanityChecks(result) + } + + test("Parse with headers without schema, inferred") { + val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, + useHeader = true, inferSchema = true) + sanityChecks(result) + } + + test("Parse with comments, ignore") { + val result = sqlContext.fixedFile(fruit_resource("comments"), fruitWidths, + useHeader = true, inferSchema = true, comment = '/') + sanityChecks(result) + } + + test("Parse malformed, schemaless, PERMISSIVE") { + val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + useHeader = false, mode = "PERMISSIVE") + result.show() + assert(result.collect().length === fruitSize) + } + + test("Parse malformed, schemaless, DROPMALFORMED") { + val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + useHeader = false, mode = "DROPMALFORMED") + result.show() + assert(result.collect().length < fruitSize) + } + + test("Parse malformed, with schema, FAILFAST") { + intercept[Exception]( + sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + fruitSchema, useHeader = false, mode = "FAILFAST").collect() + ) + } + } From a37f883e52bae9418e0f109829f566c578d6c64f Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Fri, 12 Feb 2016 14:09:49 -0500 Subject: [PATCH 4/6] Readme and code style fixes --- README.md | 13 +++++++------ .../spark/fixedwidth/FixedwidthRelation.scala | 8 +++----- .../quartethealth/spark/fixedwidth/package.scala | 3 +-- .../spark/fixedwidth/readers/readers.scala | 2 +- .../spark/fixedwidth/FixedwidthSuite.scala | 7 ++++--- 5 files changed, 16 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 97814a9..d6533c8 100644 --- a/README.md +++ b/README.md @@ -39,8 +39,9 @@ __Spark 1.4+:__ See [sample fixed-width files](src/test/resources) ```scala -import org.apache.spark.sql.{DataFrame, SQLContext} +import org.apache.spark.sql.SQLContext import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} +import com.quartethealth.spark.fixedwidth.FixedwidthContext val fruitSchema = StructType(Seq( StructField("val", IntegerType), @@ -49,9 +50,9 @@ val fruitSchema = StructType(Seq( StructField("cost", DoubleType) )) -val sqlContext = new SQLContext(sc) +val sqlContext = new SQLContext(sc) // sc is defined in the spark console val fruitWidths = Array(3, 10, 5, 4) -val fruit_resource = 'fruit_fixedwidths.txt' +val fruit_resource = "fruit_fixedwidths.txt" val result = sqlContext.fixedFile( fruit_resource, @@ -62,7 +63,7 @@ val result = sqlContext.fixedFile( result.show() // Prints top 20 rows in tabular format // Example without schema, and showing extra options -val fruit_resource = 'fruit_w_headers_fixedwidths.txt' +val fruit_resource = "fruit_w_headers_fixedwidths.txt" val result = sqlContext.fixedFile( fruit_resource, fruitWidths, @@ -70,8 +71,8 @@ val result = sqlContext.fixedFile( inferSchema = true, mode = "DROPMALFORMED", comment = '/', - ignoreLeadingWhiteSpace: true, - ignoreTrailingWhiteSpace: false, + ignoreLeadingWhiteSpace = true, + ignoreTrailingWhiteSpace = false ) result.collect() // Returns an array that contains all of Rows in this DataFrame ``` diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala index 8410f8a..2ceb083 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/FixedwidthRelation.scala @@ -1,12 +1,11 @@ package com.quartethealth.spark.fixedwidth +import com.databricks.spark.csv.CsvRelation import com.databricks.spark.csv.readers.{BulkReader, LineReader} -import com.quartethealth.spark.fixedwidth.readers.{LineFixedwidthReader, BulkFixedwidthReader} +import com.quartethealth.spark.fixedwidth.readers.{BulkFixedwidthReader, LineFixedwidthReader} import org.apache.spark.rdd.RDD -import org.apache.spark.sql.types.StructType import org.apache.spark.sql.SQLContext - -import com.databricks.spark.csv.CsvRelation +import org.apache.spark.sql.types.StructType class FixedwidthRelation protected[spark] ( baseRDD: () => RDD[String], @@ -41,7 +40,6 @@ class FixedwidthRelation protected[spark] ( protected override def getLineReader(): LineReader = { val commentChar: Char = if (comment == null) '\0' else comment - new LineFixedwidthReader(fixedWidths, commentMarker = commentChar, ignoreLeadingSpace = ignoreLeadingWhiteSpace, ignoreTrailingSpace = ignoreTrailingWhiteSpace) diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala index 7f3e50a..e14dabd 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/package.scala @@ -1,12 +1,11 @@ package com.quartethealth.spark - import com.databricks.spark.csv.util.TextFile import org.apache.spark.sql.types.StructType import org.apache.spark.sql.{DataFrame, SQLContext} package object fixedwidth { - implicit class CsvContext(sqlContext: SQLContext) extends Serializable { + implicit class FixedwidthContext(sqlContext: SQLContext) extends Serializable { def fixedFile( filePath: String, fixedWidths: Array[Int], diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala index f6843a8..e02857b 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala @@ -1,7 +1,7 @@ package com.quartethealth.spark.fixedwidth.readers import com.databricks.spark.csv.readers.{BulkReader, LineReader} -import com.univocity.parsers.fixed.{FixedWidthParserSettings, FixedWidthFieldLengths, FixedWidthParser} +import com.univocity.parsers.fixed.{FixedWidthFieldLengths, FixedWidthParser, FixedWidthParserSettings} /** * Read and parse Fixed-width-like input diff --git a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala index ba4cdf3..07ccdc8 100644 --- a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala +++ b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala @@ -1,13 +1,14 @@ package com.quartethealth.spark.fixedwidth - import org.apache.spark.SparkContext +import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType} import org.apache.spark.sql.{DataFrame, SQLContext} -import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} import org.scalatest.{BeforeAndAfterAll, FunSuite} + class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { - protected def fruit_resource(name: String = ""): String = s"src/test/resources/fruit_${name}_fixedwidth.txt" + protected def fruit_resource(name: String = ""): String = + s"src/test/resources/fruit_${name}_fixedwidth.txt" protected val fruitWidths = Array(3, 10, 5, 4) protected val fruitSize = 7 From f9b1821dd07b2d3712f028fadb9bc2a0c950e750 Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Tue, 16 Feb 2016 14:12:58 -0500 Subject: [PATCH 5/6] Added dependency on our github-hosted fork of spark-csv This means we don't need to use a JAR for the dep. --- build.sbt | 2 ++ project/Build.scala | 8 ++++++++ 2 files changed, 10 insertions(+) create mode 100644 project/Build.scala diff --git a/build.sbt b/build.sbt index 6a200bd..5e648a1 100644 --- a/build.sbt +++ b/build.sbt @@ -2,6 +2,8 @@ name := "spark-fixedwidth" version := "1.0" +organization := "com.quartethealth" + scalaVersion := "2.11.7" libraryDependencies ++= Seq( diff --git a/project/Build.scala b/project/Build.scala new file mode 100644 index 0000000..6c1f0e2 --- /dev/null +++ b/project/Build.scala @@ -0,0 +1,8 @@ +import sbt._ + +object MyBuild extends Build { + + lazy val root = Project("root", file(".")) dependsOn csvProj + lazy val csvProj = RootProject(uri("git://github.com/quartethealth/spark-csv")) + +} \ No newline at end of file From b9e60e4e6473c14c78f0b11ff8e5c7514c150959 Mon Sep 17 00:00:00 2001 From: Ben Lee Rodgers Date: Wed, 17 Feb 2016 18:34:43 -0500 Subject: [PATCH 6/6] Convert tests to specs2 format Tidied-up a few things as per the PR comments --- build.sbt | 8 +- .../spark/fixedwidth/readers/readers.scala | 4 +- .../fruit_wrong_schema_fixedwidth.txt | 7 + .../spark/fixedwidth/FixedwidthSuite.scala | 158 +++++++++--------- 4 files changed, 96 insertions(+), 81 deletions(-) create mode 100644 src/test/resources/fruit_wrong_schema_fixedwidth.txt diff --git a/build.sbt b/build.sbt index 5e648a1..3df4cf2 100644 --- a/build.sbt +++ b/build.sbt @@ -10,5 +10,9 @@ libraryDependencies ++= Seq( "com.univocity" % "univocity-parsers" % "1.5.1", "org.apache.spark" %% "spark-sql" % "1.6.0" % "provided", "org.apache.spark" %% "spark-core" % "1.6.0" % "provided", - "org.scalatest" %% "scalatest" % "2.2.1" % "test" -) \ No newline at end of file + "org.scalatest" %% "scalatest" % "2.2.1" % "test", + + "org.specs2" %% "specs2-core" % "3.7" % "test" +) + +scalacOptions in Test ++= Seq("-Yrangepos") diff --git a/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala index e02857b..d65327a 100644 --- a/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala +++ b/src/main/scala/com/quartethealth/spark/fixedwidth/readers/readers.scala @@ -79,7 +79,7 @@ private[fixedwidth] class LineFixedwidthReader( * @return array of strings where each string is a field in the CSV record */ def parseLine(line: String): Array[String] = { - parser.beginParsing(getReader(line)) + parser.beginParsing(reader(line)) val parsed = parser.parseNext() parser.stopParsing() parsed @@ -120,7 +120,7 @@ private[fixedwidth] class BulkFixedwidthReader( maxCols ) with BulkReader { - parser.beginParsing(getReader(iter)) + parser.beginParsing(reader(iter)) private var nextRecord = parser.parseNext() /** diff --git a/src/test/resources/fruit_wrong_schema_fixedwidth.txt b/src/test/resources/fruit_wrong_schema_fixedwidth.txt new file mode 100644 index 0000000..c9cdc6b --- /dev/null +++ b/src/test/resources/fruit_wrong_schema_fixedwidth.txt @@ -0,0 +1,7 @@ +so apple TRUE 0.56 +on pear FALSE1.34 +hi raspberry TRUE 2.43 +yo plum TRUE 1.31 +ni cherry TRUE 1.4 +po orange FALSE2.34 +no persimmon FALSE23.2 \ No newline at end of file diff --git a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala index 07ccdc8..8cf7a33 100644 --- a/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala +++ b/src/test/scala/com/quartethealth/spark/fixedwidth/FixedwidthSuite.scala @@ -1,17 +1,18 @@ package com.quartethealth.spark.fixedwidth -import org.apache.spark.SparkContext +import org.apache.spark.{SparkContext, SparkException} import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType} import org.apache.spark.sql.{DataFrame, SQLContext} -import org.scalatest.{BeforeAndAfterAll, FunSuite} +import org.specs2.mutable.Specification +import org.specs2.specification.After - -class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { +trait FixedwidthSetup extends After { protected def fruit_resource(name: String = ""): String = s"src/test/resources/fruit_${name}_fixedwidth.txt" protected val fruitWidths = Array(3, 10, 5, 4) protected val fruitSize = 7 + protected val malformedFruitSize = 5 protected val fruitFirstRow = Seq(56, "apple", "TRUE", 0.56) protected val fruitSchema = StructType(Seq( @@ -21,97 +22,100 @@ class FixedwidthSuite extends FunSuite with BeforeAndAfterAll { StructField("cost", DoubleType) )) - protected var sqlContext: SQLContext = _ + val sqlContext: SQLContext = new SQLContext(new SparkContext("local[2]", "FixedwidthSuite")) - override protected def beforeAll(): Unit = { - super.beforeAll() - sqlContext = new SQLContext(new SparkContext("local[2]", "FixedwidthSuite")) - } + def after = sqlContext.sparkContext.stop() +} - override protected def afterAll(): Unit = { - try { - sqlContext.sparkContext.stop() - } finally { - super.afterAll() - } - } +class FixedwidthSpec extends Specification with FixedwidthSetup { - protected def sanityChecks(resultSet: DataFrame): Unit = { + protected def sanityChecks(resultSet: DataFrame) = { resultSet.show() - assert(resultSet.collect().length === fruitSize) + resultSet.collect().length mustEqual fruitSize val head = resultSet.head() - assert(head.length === fruitWidths.length) - assert(head.toSeq === fruitFirstRow) + head.length mustEqual fruitWidths.length + head.toSeq mustEqual fruitFirstRow } - test("Parse basic") { - val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, fruitSchema, - useHeader = false) - sanityChecks(result) - } + "FixedwidthParser" should { + "Parse a basic fixed width file, successfully" in { + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, fruitSchema, + useHeader = false) + sanityChecks(result) + } - test("Parse with headers, ignore") { - val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, - fruitSchema, useHeader = true) - sanityChecks(result) - } + "Parse a fw file with headers, and ignore them" in { + val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, + fruitSchema, useHeader = true) + sanityChecks(result) + } - test("Parse with overflow, ignore") { - val result = sqlContext.fixedFile(fruit_resource("overflow"), fruitWidths, - fruitSchema, useHeader = false) - sanityChecks(result) - } + "Parse a fw file with overflowing lines, and ignore the overflow" in { + val result = sqlContext.fixedFile(fruit_resource("overflow"), fruitWidths, + fruitSchema, useHeader = false) + sanityChecks(result) + } - test("Parse with underflow, ignore") { - val result = sqlContext.fixedFile(fruit_resource("underflow"), fruitWidths, - fruitSchema, useHeader = false) - sanityChecks(result) - } + "Parse a fw file with underflowing lines, successfully " in { + val result = sqlContext.fixedFile(fruit_resource("underflow"), fruitWidths, + fruitSchema, useHeader = false) + sanityChecks(result) + } - test("Parse basic without schema, uninferred") { - val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, - useHeader = false, inferSchema = false) - sanityChecks(result) - } + "Parse a basic fw file without schema and without inferring types, successfully" in { + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, + useHeader = false, inferSchema = false) + sanityChecks(result) + } - test("Parse basic without schema, inferred") { - val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, - useHeader = false, inferSchema = true) - sanityChecks(result) - } + "Parse a basic fw file without schema, and infer the schema" in { + val result = sqlContext.fixedFile(fruit_resource(), fruitWidths, + useHeader = false, inferSchema = true) + sanityChecks(result) + } - test("Parse with headers without schema, inferred") { - val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, - useHeader = true, inferSchema = true) - sanityChecks(result) - } + "Parse a fw file with headers but without schema and without inferrence, succesfully" in { + val result = sqlContext.fixedFile(fruit_resource("w_headers"), fruitWidths, + useHeader = true, inferSchema = true) + sanityChecks(result) + } - test("Parse with comments, ignore") { - val result = sqlContext.fixedFile(fruit_resource("comments"), fruitWidths, - useHeader = true, inferSchema = true, comment = '/') - sanityChecks(result) - } + "Parse a fw file with comments, and ignore those lines" in { + val result = sqlContext.fixedFile(fruit_resource("comments"), fruitWidths, + useHeader = true, inferSchema = true, comment = '/') + sanityChecks(result) + } - test("Parse malformed, schemaless, PERMISSIVE") { - val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, - useHeader = false, mode = "PERMISSIVE") - result.show() - assert(result.collect().length === fruitSize) - } + "Parse a malformed fw and schemaless file in PERMISSIVE mode, successfully" in { + val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + useHeader = false, mode = "PERMISSIVE") + result.show() + result.collect().length mustEqual fruitSize + } - test("Parse malformed, schemaless, DROPMALFORMED") { - val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, - useHeader = false, mode = "DROPMALFORMED") - result.show() - assert(result.collect().length < fruitSize) - } + "Parse a malformed and schemaless fw file in DROPMALFORMED mode, successfully dropping bad lines" in { + val result = sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + useHeader = false, mode = "DROPMALFORMED") + result.show() + result.collect().length mustEqual malformedFruitSize + } - test("Parse malformed, with schema, FAILFAST") { - intercept[Exception]( - sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, - fruitSchema, useHeader = false, mode = "FAILFAST").collect() - ) + "FAIL to parse a malformed fw file with schema in FAILFAST mode" in { + def fail = { + sqlContext.fixedFile(fruit_resource("malformed"), fruitWidths, + fruitSchema, useHeader = false, mode = "FAILFAST").collect() + } + fail must throwA[SparkException] + } + + "FAIL to parse a fw file with the wrong format" in { + def fail = { + sqlContext.fixedFile(fruit_resource("wrong_schema"), fruitWidths, + fruitSchema, useHeader = false).collect() + } + fail must throwA[SparkException] + } } }