加载csv文件时获取com.univocity.parsers.common.TextParsingException

Question

我正在尝试加入一个tsv数据集，它在数据中有很多新行到另一个dataframe，并不断地获得

com.univocity.parsers.common.TextParsingException

我已经清理了我的数据，将\N替换为NAs，因为我认为这可能是原因，但没有成功。

该错误指向错误数据中的以下记录

tt0100054 2ПовелительмухSUHH ru NA NA 0

堆栈跟踪如下所示

    19/03/02 17:45:42 ERROR Executor: Exception in task 0.0 in stage 10.0 (TID 10)
com.univocity.parsers.common.TextParsingException: Length of parsed input (1000001) exceeds the maximum number of characters defined in your parser settings (1000000). 
Identified line separator characters in the parsed content. This may be the cause of the error. The line separator in your parser settings is set to '\n'. Parsed content:
    Sesso e il poliziotto sposato   IT  NA  NA  NA  0[\n]
    tt0097089   4   Sex and the Married Detective   US  NA  NA  NA  0[\n]`tt0100054 1   Fluenes herre   NO  NA  imdbDisplay NA  0
tt0100054   20  Kärpästen herra FI  NA  NA  NA  0
tt0100054   2
    at com.univocity.parsers.common.AbstractParser.handleException(AbstractParser.java:302)
    at com.univocity.parsers.common.AbstractParser.parseNext(AbstractParser.java:431)
    at org.apache.spark.sql.execution.datasources.csv.BulkCsvReader.next(CSVParser.scala:148)
    at org.apache.spark.sql.execution.datasources.csv.BulkCsvReader.next(CSVParser.scala:131)
    at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:91)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:246)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:240)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
    at org.apache.spark.scheduler.Task.run(Task.scala:86)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 1000000
    at com.univocity.parsers.common.input.AbstractCharInputReader.appendUtilAnyEscape(AbstractCharInputReader.java:331)
    at com.univocity.parsers.csv.CsvParser.parseQuotedValue(CsvParser.java:246)
    at com.univocity.parsers.csv.CsvParser.parseRecord(CsvParser.java:119)
    at com.univocity.parsers.common.AbstractParser.parseNext(AbstractParser.java:400)
    ... 22 more

我已经尝试过在csv选项(“maxCharsPerCol”、"110000000")中设置.option("multiLine“、"true")，但这没有帮助。如果能帮我解决这个问题我会很感激的。

我使用的是spark 2.0.2和Scala2.11.8。

Answer 1

univocity-parsers的作者。

当您的程序(即文件格式没有正确配置)或输入文件(即输入文件没有您的程序所期望的格式，或者有未转义/未关闭的引号)可能出现问题时，解析器就会快速失败。

堆栈跟踪显示如下：

Sesso e il poliziotto sposato   IT  NA  NA  NA  0[\n]
tt0097089   4   Sex and the Married Detective   US  NA  NA  NA  0[\n]`tt0100054 1   Fluenes herre   NO  NA  imdbDisplay NA  0
tt0100054   20  Kärpästen herra FI  NA  NA  NA  0
tt0100054   2

它清楚地显示被读取的多行的内容，就好像它们是单个值的一部分一样。这意味着，在输入文件中的某个地方，有一些值以引号开头，这些值永远不会关闭。

您可以将解析器配置为不尝试使用以下方法处理引用的值：

settings.getFormat().setQuote('\0');

如果您确信您的格式配置是正确的，并且输入中有很长的值，请将maxCharsPerColumn设置为-1。

最后，看起来您正在解析TSV，它不是CSV，应该进行不同的处理。如果是这样的话，您也可以尝试使用TsvParser。

希望这能有所帮助

Answer 2

杰罗尼莫的回答将解决这个问题。

只是添加一个示例代码块，以防您想知道如何做到这一点。

val tsvData = spark.read.option("header","true").option("inferSchema",
"true").option("delimiter","\t").option("quote","\0").csv(csvFilePath)

Answer 3

有关任何人遇到此问题，请参阅https://spark.apache.org/docs/latest/sql-data-sources-csv.html

星火中的CSV读取器有一个设置maxColumns，它被设置为20480 (从Spark3.3开始)。

通过将其设置为至少与预期列数相同的数字(如果已知的话)，可以增加此限制：

spark.read.format("csv").option("header", "true").option("maxColumns", 500000).load(filename)

请记住，增加maxColumns是一种权衡--您预先分配了更多的内存，因此在某个时候，您会因为预先分配过多的额外空间而耗尽内存。