Skip to content

[SPARK-18699][SQL] Put malformed tokens into a new field when parsing CSV data #16928

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

Closed
maropu wants to merge 12 commits into from
Closed

[SPARK-18699][SQL] Put malformed tokens into a new field when parsing CSV data #16928

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
74e1fc5
Fill NULL in a field when detecting a malformed token
maropu Feb 14, 2017
763601d
Add an option columnNameOfCorruptRecord for CSV formats
maropu Feb 16, 2017
873a383
Apply review comments
maropu Feb 18, 2017
4eed4a4
Apply review comments
maropu Feb 20, 2017
448e6fe
Apply review comments
maropu Feb 22, 2017
619094a
Add more tests
maropu Feb 22, 2017
80c3775
Replace map with foreach
maropu Feb 23, 2017
c86febe
Apply review comments
maropu Feb 23, 2017
8d9386a
Update docs
maropu Feb 23, 2017
512fb42
Clean up code in UnivocityParser
maropu Feb 23, 2017
3d514e5
Revert clean-up code for a CSV parser
maropu Feb 23, 2017
a58ff1f
Update descriptions for comments
maropu Feb 23, 2017
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
24 changes: 18 additions & 6 deletions python/pyspark/sql/readwriter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,8 +193,9 @@ def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,

* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
record and puts the malformed string into a new field configured by \
``columnNameOfCorruptRecord``. When a schema is set by user, it sets \
``null`` for extra fields.
``columnNameOfCorruptRecord``. An user-defined schema can include \

@cloud-fan cloud-fan Feb 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please rephrase this document a little bit, to make it more clear

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay, I'll brush up

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated and could you check this again? Thanks!

a string type field named ``columnNameOfCorruptRecord`` for corrupt records. \
When a schema is set by user, it sets ``null`` for extra fields.

@cloud-fan cloud-fan Feb 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about the other 2 modes? do they also set null for extra fields?

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. The two mode does not set null. In failFast mode, it fails a job. In dropMalformed mode, it drops the malformed lines whose length is shorter or longer.

@cloud-fan cloud-fan Feb 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does json have similar behavior?

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah..., a bit different I think. As @HyukjinKwon said above(#16928 (comment)), CSV formats depend on a length of parsed tokens (if the length shorter, fills null, and if longer, drops them in permissive mode). One the other hand, in JSON formats, fields in a required schema are mapped by key. In case of missing keys in JSON formats, it just sets null in these fields with the keys in all the three mode. cc: @HyukjinKwon
e.x.)


import org.apache.spark.sql.types._
scala> Seq("""{"a": "a", "b" : 1}""", """{"a": "a"}""").toDF().write.text("/Users/maropu/Desktop/data")
scala> val dataSchema = StructType(StructField("a", StringType, true) :: StructField("b", IntegerType, true) :: Nil)
scala> spark.read.schema(dataSchema).option("mode", "PERMISSIVE").json("/Users/maropu/Desktop/data").show()
+---+----+
|  a|   b|
+---+----+
|  a|   1|
|  a|null|
+---+----+

scala> spark.read.schema(dataSchema).option("mode", "FAILFAST").json("/Users/maropu/Desktop/data").show()
+---+----+
|  a|   b|
+---+----+
|  a|   1|
|  a|null|
+---+----+

scala> spark.read.schema(dataSchema).option("mode", "DROPMALFORMED").json("/Users/maropu/Desktop/data").show()
+---+----+
|  a|   b|
+---+----+
|  a|   1|
|  a|null|
+---+----+

* ``DROPMALFORMED`` : ignores the whole corrupted records.
* ``FAILFAST`` : throws an exception when it meets corrupted records.

Expand Down Expand Up @@ -304,7 +305,8 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
comment=None, header=None, inferSchema=None, ignoreLeadingWhiteSpace=None,
ignoreTrailingWhiteSpace=None, nullValue=None, nanValue=None, positiveInf=None,
negativeInf=None, dateFormat=None, timestampFormat=None, maxColumns=None,
maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, timeZone=None):
maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, timeZone=None,
columnNameOfCorruptRecord=None):

@HyukjinKwon HyukjinKwon Feb 22, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doh, it seems we should add this in streaming.py and DataStreamReader too.

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay, I'll check soon

"""Loads a CSV file and returns the result as a :class:`DataFrame`.

This function will go through the input once to determine the input schema if
Expand Down Expand Up @@ -366,11 +368,20 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
:param timeZone: sets the string that indicates a timezone to be used to parse timestamps.
If None is set, it uses the default value, session local timezone.

* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted record.
When a schema is set by user, it sets ``null`` for extra fields.
* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
record and puts the malformed string into a new field configured by \
``columnNameOfCorruptRecord``. An user-defined schema can include \
a string type field named ``columnNameOfCorruptRecord`` for corrupt records. \
When a schema is set by user, it sets ``null`` for extra fields.
* ``DROPMALFORMED`` : ignores the whole corrupted records.
* ``FAILFAST`` : throws an exception when it meets corrupted records.

:param columnNameOfCorruptRecord: allows renaming the new field having malformed string
created by ``PERMISSIVE`` mode. This overrides
``spark.sql.columnNameOfCorruptRecord``. If None is set,
it uses the value specified in
``spark.sql.columnNameOfCorruptRecord``.

>>> df = spark.read.csv('python/test_support/sql/ages.csv')
>>> df.dtypes
[('_c0', 'string'), ('_c1', 'string')]
Expand All @@ -382,7 +393,8 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
nanValue=nanValue, positiveInf=positiveInf, negativeInf=negativeInf,
dateFormat=dateFormat, timestampFormat=timestampFormat, maxColumns=maxColumns,
maxCharsPerColumn=maxCharsPerColumn,
maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode, timeZone=timeZone)
maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode, timeZone=timeZone,
columnNameOfCorruptRecord=columnNameOfCorruptRecord)
if isinstance(path, basestring):
path = [path]
return self._df(self._jreader.csv(self._spark._sc._jvm.PythonUtils.toSeq(path)))
Expand Down
24 changes: 18 additions & 6 deletions python/pyspark/sql/streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,8 +465,9 @@ def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,

* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
record and puts the malformed string into a new field configured by \
``columnNameOfCorruptRecord``. When a schema is set by user, it sets \
``null`` for extra fields.
``columnNameOfCorruptRecord``. An user-defined schema can include \
a string type field named ``columnNameOfCorruptRecord`` for corrupt records. \
When a schema is set by user, it sets ``null`` for extra fields.
* ``DROPMALFORMED`` : ignores the whole corrupted records.
* ``FAILFAST`` : throws an exception when it meets corrupted records.

Expand Down Expand Up @@ -558,7 +559,8 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
comment=None, header=None, inferSchema=None, ignoreLeadingWhiteSpace=None,
ignoreTrailingWhiteSpace=None, nullValue=None, nanValue=None, positiveInf=None,
negativeInf=None, dateFormat=None, timestampFormat=None, maxColumns=None,
maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, timeZone=None):
maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, timeZone=None,
columnNameOfCorruptRecord=None):
"""Loads a CSV file stream and returns the result as a :class:`DataFrame`.

This function will go through the input once to determine the input schema if
Expand Down Expand Up @@ -618,11 +620,20 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
:param timeZone: sets the string that indicates a timezone to be used to parse timestamps.
If None is set, it uses the default value, session local timezone.

* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted record.
When a schema is set by user, it sets ``null`` for extra fields.
* ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
record and puts the malformed string into a new field configured by \
``columnNameOfCorruptRecord``. An user-defined schema can include \
a string type field named ``columnNameOfCorruptRecord`` for corrupt records. \
When a schema is set by user, it sets ``null`` for extra fields.
* ``DROPMALFORMED`` : ignores the whole corrupted records.
* ``FAILFAST`` : throws an exception when it meets corrupted records.

:param columnNameOfCorruptRecord: allows renaming the new field having malformed string
created by ``PERMISSIVE`` mode. This overrides
``spark.sql.columnNameOfCorruptRecord``. If None is set,
it uses the value specified in
``spark.sql.columnNameOfCorruptRecord``.

>>> csv_sdf = spark.readStream.csv(tempfile.mkdtemp(), schema = sdf_schema)
>>> csv_sdf.isStreaming
True
Expand All @@ -636,7 +647,8 @@ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=Non
nanValue=nanValue, positiveInf=positiveInf, negativeInf=negativeInf,
dateFormat=dateFormat, timestampFormat=timestampFormat, maxColumns=maxColumns,
maxCharsPerColumn=maxCharsPerColumn,
maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode, timeZone=timeZone)
maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode, timeZone=timeZone,
columnNameOfCorruptRecord=columnNameOfCorruptRecord)
if isinstance(path, basestring):
return self._df(self._jreader.csv(path))
else:
Expand Down
14 changes: 10 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,8 +286,9 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
* during parsing.
* <ul>
* <li>`PERMISSIVE` : sets other fields to `null` when it meets a corrupted record, and puts
* the malformed string into a new field configured by `columnNameOfCorruptRecord`. When
* a schema is set by user, it sets `null` for extra fields.</li>
* the malformed string into a new field configured by `columnNameOfCorruptRecord`.
* An user-defined schema can include a string type field named `columnNameOfCorruptRecord`
* for corrupt records. When a schema is set by user, it sets `null` for extra fields.</li>
* <li>`DROPMALFORMED` : ignores the whole corrupted records.</li>
* <li>`FAILFAST` : throws an exception when it meets corrupted records.</li>
* </ul>
Expand Down Expand Up @@ -422,12 +423,17 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
* <li>`mode` (default `PERMISSIVE`): allows a mode for dealing with corrupt records
* during parsing.
* <ul>
* <li>`PERMISSIVE` : sets other fields to `null` when it meets a corrupted record. When
* a schema is set by user, it sets `null` for extra fields.</li>
* <li>`PERMISSIVE` : sets other fields to `null` when it meets a corrupted record, and puts
* the malformed string into a new field configured by `columnNameOfCorruptRecord`.
* An user-defined schema can include a string type field named `columnNameOfCorruptRecord`
* for corrupt records. When a schema is set by user, it sets `null` for extra fields.</li>
* <li>`DROPMALFORMED` : ignores the whole corrupted records.</li>
* <li>`FAILFAST` : throws an exception when it meets corrupted records.</li>
* </ul>
* </li>
* <li>`columnNameOfCorruptRecord` (default is the value specified in
* `spark.sql.columnNameOfCorruptRecord`): allows renaming the new field having malformed string
* created by `PERMISSIVE` mode. This overrides `spark.sql.columnNameOfCorruptRecord`.</li>
* </ul>
* @since 2.0.0
*/
Expand Down
31 changes: 22 additions & 9 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,9 @@ import org.apache.hadoop.mapreduce._

import org.apache.spark.TaskContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
import org.apache.spark.sql.{AnalysisException, Dataset, Encoders, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CompressionCodecs}
import org.apache.spark.sql.catalyst.util.CompressionCodecs
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.text.TextFileFormat
import org.apache.spark.sql.sources._
Expand Down Expand Up @@ -96,31 +96,44 @@ class CSVFileFormat extends TextBasedFileFormat with DataSourceRegister {
filters: Seq[Filter],
options: Map[String, String],
hadoopConf: Configuration): (PartitionedFile) => Iterator[InternalRow] = {
val csvOptions = new CSVOptions(options, sparkSession.sessionState.conf.sessionLocalTimeZone)

CSVUtils.verifySchema(dataSchema)
val broadcastedHadoopConf =
sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))

val parsedOptions = new CSVOptions(
options,
sparkSession.sessionState.conf.sessionLocalTimeZone,
sparkSession.sessionState.conf.columnNameOfCorruptRecord)

@HyukjinKwon HyukjinKwon Feb 19, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(It seems CSVOptions is created twice above :)).

@maropu maropu Feb 20, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed


// Check a field requirement for corrupt records here to throw an exception in a driver side
dataSchema.getFieldIndex(parsedOptions.columnNameOfCorruptRecord).foreach { corruptFieldIndex =>
val f = dataSchema(corruptFieldIndex)
if (f.dataType != StringType || !f.nullable) {

@HyukjinKwon HyukjinKwon Feb 22, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This check seems duplicated with https://github.com/apache/spark/pull/16928/files/4eed4a4bea927c6365dac2e2734c895a0a1a0026#diff-d19881aceddcaa5c60620fdcda99b4c4R51

@maropu maropu Feb 22, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I remove the entry in UnivocityParser.

throw new AnalysisException(
"The field for corrupt records must be string type and nullable")
}
}

(file: PartitionedFile) => {
val lines = {
val conf = broadcastedHadoopConf.value.value
val linesReader = new HadoopFileLinesReader(file, conf)
Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => linesReader.close()))
linesReader.map { line =>
new String(line.getBytes, 0, line.getLength, csvOptions.charset)
new String(line.getBytes, 0, line.getLength, parsedOptions.charset)
}
}

val linesWithoutHeader = if (csvOptions.headerFlag && file.start == 0) {
val linesWithoutHeader = if (parsedOptions.headerFlag && file.start == 0) {
// Note that if there are only comments in the first block, the header would probably
// be not dropped.
CSVUtils.dropHeaderLine(lines, csvOptions)
CSVUtils.dropHeaderLine(lines, parsedOptions)
} else {
lines
}

val filteredLines = CSVUtils.filterCommentAndEmpty(linesWithoutHeader, csvOptions)
val parser = new UnivocityParser(dataSchema, requiredSchema, csvOptions)
val filteredLines = CSVUtils.filterCommentAndEmpty(linesWithoutHeader, parsedOptions)
val parser = new UnivocityParser(dataSchema, requiredSchema, parsedOptions)
filteredLines.flatMap(parser.parse)
}
}
Expand Down
18 changes: 15 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVOptions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,20 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CompressionCodecs, ParseModes}

private[csv] class CSVOptions(
@transient private val parameters: CaseInsensitiveMap[String], defaultTimeZoneId: String)
@transient private val parameters: CaseInsensitiveMap[String],
defaultTimeZoneId: String,
defaultColumnNameOfCorruptRecord: String)
extends Logging with Serializable {

def this(parameters: Map[String, String], defaultTimeZoneId: String) =
this(CaseInsensitiveMap(parameters), defaultTimeZoneId)
def this(
parameters: Map[String, String],
defaultTimeZoneId: String,
defaultColumnNameOfCorruptRecord: String = "") = {
this(
CaseInsensitiveMap(parameters),
defaultTimeZoneId,
defaultColumnNameOfCorruptRecord)
}

private def getChar(paramName: String, default: Char): Char = {
val paramValue = parameters.get(paramName)
Expand Down Expand Up @@ -95,6 +104,9 @@ private[csv] class CSVOptions(
val dropMalformed = ParseModes.isDropMalformedMode(parseMode)
val permissive = ParseModes.isPermissiveMode(parseMode)

val columnNameOfCorruptRecord =
parameters.getOrElse("columnNameOfCorruptRecord", defaultColumnNameOfCorruptRecord)

@HyukjinKwon HyukjinKwon Feb 19, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe, we should add this in readwriter.py too and document this in readwriter.py, DataFrameReader and DataStreamReader.

@maropu maropu Feb 20, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added doc descriptions in readwriter.py, DataFrameReader, and DataStreamReader .


val nullValue = parameters.getOrElse("nullValue", "")

val nanValue = parameters.getOrElse("nanValue", "NaN")
Expand Down
46 changes: 38 additions & 8 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,14 @@ private[csv] class UnivocityParser(
// A `ValueConverter` is responsible for converting the given value to a desired type.
private type ValueConverter = String => Any

private val corruptFieldIndex = schema.getFieldIndex(options.columnNameOfCorruptRecord)
corruptFieldIndex.foreach { corrFieldIndex =>
require(schema(corrFieldIndex).dataType == StringType)
require(schema(corrFieldIndex).nullable)
}

private val dataSchema = StructType(schema.filter(_.name != options.columnNameOfCorruptRecord))

@HyukjinKwon HyukjinKwon Feb 23, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just realised now we only use the length of dataSchema now. Could we just use the length if more commits should be pushed?

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I'll update

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I reverted some parts of code and then dataSchema is used except for the length https://github.com/apache/spark/pull/16928/files#diff-d19881aceddcaa5c60620fdcda99b4c4R57. So, I kept this variable as it is.


private val valueConverters =
schema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray

Expand Down Expand Up @@ -148,6 +156,7 @@ private[csv] class UnivocityParser(
case udt: UserDefinedType[_] => (datum: String) =>
makeConverter(name, udt.sqlType, nullable, options)

// We don't actually hit this exception though, we keep it for understandability
case _ => throw new RuntimeException(s"Unsupported type: ${dataType.typeName}")
}

Expand All @@ -172,7 +181,7 @@ private[csv] class UnivocityParser(
* the record is malformed).
*/
def parse(input: String): Option[InternalRow] = {
convertWithParseMode(parser.parseLine(input)) { tokens =>
convertWithParseMode(input) { tokens =>
var i: Int = 0
while (i < indexArr.length) {
val pos = indexArr(i)
Expand All @@ -190,8 +199,9 @@ private[csv] class UnivocityParser(
}

private def convertWithParseMode(
tokens: Array[String])(convert: Array[String] => InternalRow): Option[InternalRow] = {
if (options.dropMalformed && schema.length != tokens.length) {
input: String)(convert: Array[String] => InternalRow): Option[InternalRow] = {
val tokens = parser.parseLine(input)
if (options.dropMalformed && dataSchema.length != tokens.length) {
if (numMalformedRecords < options.maxMalformedLogPerPartition) {
logWarning(s"Dropping malformed line: ${tokens.mkString(options.delimiter.toString)}")
}
Expand All @@ -202,21 +212,41 @@ private[csv] class UnivocityParser(
}
numMalformedRecords += 1
None
} else if (options.failFast && schema.length != tokens.length) {
} else if (options.failFast && dataSchema.length != tokens.length) {
throw new RuntimeException(s"Malformed line in FAILFAST mode: " +
s"${tokens.mkString(options.delimiter.toString)}")
} else {
val checkedTokens = if (options.permissive && schema.length > tokens.length) {
tokens ++ new Array[String](schema.length - tokens.length)
} else if (options.permissive && schema.length < tokens.length) {
tokens.take(schema.length)
val checkedTokens = if (options.permissive) {
// If a length of parsed tokens is not equal to expected one, it makes the length the same
// with the expected. If the length is shorter, it adds extra tokens in the tail.
// If longer, it drops extra tokens.

@cloud-fan cloud-fan Feb 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should revisit this in the future. If the token length doesn't match the expected schema, we should treat it as a malformed record. cc @HyukjinKwon

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we also put that malformed record (shorter or longer) into a corrupt field?

@HyukjinKwon HyukjinKwon Feb 23, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yup, I agree in a way but I guess "it is pretty common that CSV is malformed in this way" (said by the analysis team in my company). Could we leave it as is for now here?

Let me try to raise a different JIRA after checking R's read.csv or other libraries.

val lengthSafeTokens = if (dataSchema.length > tokens.length) {
tokens ++ new Array[String](dataSchema.length - tokens.length)
} else if (dataSchema.length < tokens.length) {
tokens.take(dataSchema.length)
} else {
tokens
}

// If we need to handle corrupt fields, it adds an extra token to skip a field for malformed

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@HyukjinKwon This fix satisfies your intention? I slightly modified code based on your code.

// strings when loading parsed tokens into a resulting `row`.
corruptFieldIndex.map { corrFieldIndex =>
val (front, back) = lengthSafeTokens.splitAt(corrFieldIndex)
front ++ new Array[String](1) ++ back

@cloud-fan cloud-fan Feb 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this will introduce a lot of extra object allocation, I think the previous version is better

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have two options; 1) we just revert this part, or (2) modify this part to avoid the allocation based on this code. cc: @HyukjinKwon

e.x.)
This is just an example and it seems to be a little hard to understand.

  val parsedTokens = new Array[String](schema.length)

        ...
        // If we need to handle corrupt fields, it adds an extra token to skip a field for malformed
        // strings when loading parsed tokens into a resulting `row`.
        corruptFieldIndex.map { corrFieldIndex =>
          lengthSafeTokens.splitAt(corrFieldIndex) match { case (front, back) =>
              front.zipWithIndex.foreach { case (s, i) =>
                parsedTokens(i) = s
              }
              back.zipWithIndex.foreach { case (s, i) =>
                parsedTokens(schema.length - back.length + i) = s
              }
          }
          parsedTokens

@HyukjinKwon HyukjinKwon Feb 23, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We value on the committer's opinion. I am fine if we revert. I personally prefer 1) revert this change then if this sounds not good.

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably, it'd be better to leave comments here as TODO.

@maropu maropu Feb 23, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added TODO comments in here and here

}.getOrElse {
lengthSafeTokens
}
} else {
tokens
}

try {
Some(convert(checkedTokens))
} catch {
case NonFatal(e) if options.permissive =>
val row = new GenericInternalRow(requiredSchema.length)
corruptFieldIndex.foreach(row(_) = UTF8String.fromString(input))
Some(row)
case NonFatal(e) if options.dropMalformed =>
if (numMalformedRecords < options.maxMalformedLogPerPartition) {
logWarning("Parse exception. " +
Expand Down
Loading