Spark CSV 包无法处理字段中的 \n

Question

我有一个 CSV 文件，我正在尝试使用 Spark CSV package 加载它，但它无法正确加载数据，因为其中很少有字段包含 \n，例如以下两行

"XYZ", "Test Data", "TestNew\nline", "OtherData" 
"XYZ", "Test Data", "blablablabla
\nblablablablablalbal", "OtherData" 

我正在使用以下代码，这很简单我在互联网上使用parserLib 作为univocity 它解决了多个换行问题，但对我来说似乎并非如此。

 SQLContext sqlContext = new SQLContext(sc);
    DataFrame df = sqlContext.read()
        .format("com.databricks.spark.csv")
        .option("inferSchema", "true")
        .option("header", "true")
        .option("parserLib","univocity")
        .load("data.csv");

如何在以引号开头的字段中替换换行符。有没有更简单的方法？

Answer 1

Spark 2.2 的用户可以使用一个选项来说明 CSV 文件中的换行符。它最初被称为wholeFile，但在发布之前更名为multiLine。

以下是使用该选项将 CSV 加载到数据框的示例：

var webtrends_data = (sparkSession.read
.option("header", "true")
.option("inferSchema", "true")
.option("multiLine", true)
.option("delimiter", ",")
.format("csv")
.load("hdfs://hadoop-master:9000/datasource/myfile.csv"))

Answer 2

根据SPARK-14194（已解决为重复）不支持带有换行符的字段，并且永远不会支持。

我建议通过wholeFile 选项解决这个问题，它似乎合并了。我正在解决这个问题，因为它有一个 PR。

然而，那是 Spark 2.0，你使用 spark-csv 模块。

在引用的SPARK-19610 中，它被pull request 修复：

嗯，我理解这样做的动机，尽管我对 csv 的理解通常避免在字段中使用换行符，或者某些实现需要使用换行符在字段值周围加上引号

换句话说，在 Spark 2.x 中使用wholeFile 选项（如您在CSVDataSource 中所见）。

至于 spark-csv，this comment 可能会有所帮助（突出显示我的）：

但是，有很多类似的 JIRA 抱怨这一点，原始 CSV 数据源试图支持这一点，尽管实现不正确。这至少会尝试将其与 JSON 匹配，并且提供一种处理此类 CSV 文件的方法可能会更好。 实际上，当前的实现需要引号 :)。 （据说 R 实际上也支持这种情况）。

在 spark-csv 的 Features 中，您可以找到以下内容：

该包还支持保存简单（非嵌套）DataFrame。写入文件时，API 接受几个选项：

• quote：默认情况下，引号字符为"，但可以设置为任何字符。这是根据quoteMode写的。

• quoteMode：何时引用字段（ALL、MINIMAL（默认）、NON_NUMERIC、NONE），请参阅引用模式

Answer 3

升级到 Spark 2.x。换行符实际上是由 ascii 13 和 10 表示的 CRLF。但是反斜杠和 'n' 是不同的 ascii，它们是通过程序解释和编写的。 Spark 2.x 将正确读取.. 我试过了..s.b。
val conf = new SparkConf().setAppName("HelloSpark").setMaster("local[2]") val sc = SparkSession.builder().master("local").getOrCreate() val df = sc.read.csv("src/main/resources/data.csv") df.foreach(row => println(row.mkString(", ")))
如果您无法升级，请使用正则表达式对 RDD 上的 \n 进行清理。这不会删除行尾，因为它是 $ 正则表达式。 S.b.

  val conf = new SparkConf().setAppName("HelloSpark").setMaster("local")
  val sc = new SparkContext(conf)
  val rdd1 = sc.textFile("src/main/resources/data.csv")
  val rdd2 = rdd1.map(row => row.replace("\\n", ""))
  val sqlContext = new SQLContext(sc)

  import sqlContext.implicits._
  val df = rdd2.toDF()
  df.foreach(row => println(row.mkString(", ")))