Skip to content

report the error as well as the line that caused it #258

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

report the error as well as the line that caused it #258

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9ddecc1
report the error as well as the line that caused it
abridgett Feb 9, 2016
09a3b1a
wrap long line
abridgett Feb 11, 2016
ebfced9
improve error message formatting (thanks to Hyukjin Kwon)
abridgett Feb 29, 2016
6ddab09
missed the all important "s" before the string quote
abridgett Mar 11, 2016
eadadd0
add wrong length information
abridgett Mar 11, 2016
72b5dec
removed trailing space
abridgett Mar 11, 2016
75e5fd3
corrected test for improved parse messages
abridgett Mar 11, 2016
382a1e1
fix long line
abridgett Mar 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 26 additions & 13 deletions src/main/scala/com/databricks/spark/csv/CsvRelation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,10 +105,14 @@ case class CsvRelation protected[spark] (
tokenRdd(schemaFields.map(_.name)).flatMap { tokens =>

if (dropMalformed && schemaFields.length != tokens.size) {
logger.warn(s"Dropping malformed line: ${tokens.mkString(",")}")
logger.warn(s"Dropping malformed line " +
s"(expected ${schemaFields.length} tokens but received " +
s"${tokens.size} tokens): ${tokens.mkString(",")}")
None
} else if (failFast && schemaFields.length != tokens.size) {
throw new RuntimeException(s"Malformed line in FAILFAST mode: ${tokens.mkString(",")}")
throw new RuntimeException(s"Malformed line " +
s"(expected ${schemaFields.length} tokens but received " +
s"${tokens.size} tokens) in FAILFAST mode: ${tokens.mkString(",")}")
} else {
var index: Int = 0
val rowArray = new Array[Any](schemaFields.length)
Expand All @@ -125,13 +129,16 @@ case class CsvRelation protected[spark] (
case aiob: ArrayIndexOutOfBoundsException if permissive =>
(index until schemaFields.length).foreach(ind => rowArray(ind) = null)
Some(Row.fromSeq(rowArray))
case _: java.lang.NumberFormatException |
_: IllegalArgumentException if dropMalformed =>
logger.warn("Number format exception. " +
case e: java.lang.NumberFormatException if dropMalformed =>
logger.warn(s"Number format exception (${e.getMessage})." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
case e: IllegalArgumentException if dropMalformed =>
logger.warn(s"IllegalArgument exception (${e.getMessage})." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
case pe: java.text.ParseException if dropMalformed =>
logger.warn("Parse exception. " +
logger.warn(s"Parse exception. (${pe.getMessage})." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
}
Expand Down Expand Up @@ -170,11 +177,14 @@ case class CsvRelation protected[spark] (
val requiredSize = requiredFields.length
tokenRdd(schemaFields.map(_.name)).flatMap { tokens =>
if (dropMalformed && schemaFields.length != tokens.size) {
logger.warn(s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
logger.warn(s"Dropping malformed line " +
s"(expected ${schemaFields.length} tokens but received " +
s"${tokens.size} tokens): ${tokens.mkString(delimiter.toString)}")
None
} else if (failFast && schemaFields.length != tokens.size) {
throw new RuntimeException(s"Malformed line in FAILFAST mode: " +
s"${tokens.mkString(delimiter.toString)}")
throw new RuntimeException(s"Malformed line in FAILFAST mode " +
s"(expected ${schemaFields.length} tokens but received " +
s"${tokens.size} tokens): ${tokens.mkString(delimiter.toString)}")
} else {
val indexSafeTokens = if (permissive && schemaFields.length != tokens.size) {
tokens ++ new Array[String](schemaFields.length - tokens.size)
Expand All @@ -198,13 +208,16 @@ case class CsvRelation protected[spark] (
}
Some(Row.fromSeq(rowArray.take(requiredSize)))
} catch {
case _: java.lang.NumberFormatException |
_: IllegalArgumentException if dropMalformed =>
logger.warn("Number format exception. " +
case e: java.lang.NumberFormatException if dropMalformed =>
logger.warn("Number format exception (" + e + ")." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
case e: IllegalArgumentException if dropMalformed =>
logger.warn("IllegalArgument exception (" + e + ")." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
case pe: java.text.ParseException if dropMalformed =>
logger.warn("Parse exception. " +
logger.warn("Parse exception (" + pe + ")." +
s"Dropping malformed line: ${tokens.mkString(delimiter.toString)}")
None
}
Expand Down
3 changes: 2 additions & 1 deletion src/test/scala/com/databricks/spark/csv/CsvSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,8 @@ abstract class AbstractCsvSuite extends FunSuite with BeforeAndAfterAll {
.collect()
}

assert(exception.getMessage.contains("Malformed line in FAILFAST mode: 2015,Chevy,Volt"))
assert(exception.getMessage.contains("Malformed line in FAILFAST mode " +
"(expected 5 tokens but received 3 tokens): 2015,Chevy,Volt"))
}

test("DSL test roundtrip nulls") {
Expand Down