Spark 2.2.0 读取csv单元格内双引号多行数据的问题

[lqshow]

Overview

Spark 2.2.0以下版本对于读取csv单元格内存在多行值(LF)是存在问题的，spark 2.2.0版本虽然修复了该问题，添加了multiLine参数，但是加上该参数后，encoding参数会失效，对于中文非utf8编码来说读取出来是一堆乱码。

Issue

• spark2.2.0 reading CSV with multiLine invalidates encoding option
• Rename wholeFile to multiLine for both CSV and JSON
• Support parsing multiline CSV files

Solution

• 使用 newAPIHadoopFile && CSVInputFormat

Apache crunch maven dependency

<dependency>
  <groupId>org.apache.crunch</groupId>
  <artifactId>crunch-core</artifactId>
  <version>0.15.0</version>
</dependency>

Configuration conf = new Configuration();
conf.set("csv.inputfileencoding", "gb18030");

jsc.newAPIHadoopFile("/Users/linqiong/Downloads/bb_gbk.csv",
  CSVInputFormat.class, null, null, conf)
  .map(s -> s._2().toString());

• 通过读取二进制文件来解决

opencsv maven dependency

<dependency>
  <groupId>com.opencsv</groupId>
  <artifactId>opencsv</artifactId>
  <version>4.1</version>
</dependency>

JavaRDD<Row> rowRDD = jsc.binaryFiles("/Users/linqiong/Downloads/bb_gbk.csv")
    .flatMap(line -> {
        PortableDataStream ds = line._2();
        DataInputStream dis = ds.open();
        List<String[]> data = new ArrayList<>();
        try (CSVReader reader = new CSVReader(new BufferedReader(new InputStreamReader(dis, GB18030)))) {
            String[] nextLine;
            while ((nextLine = reader.readNext()) != null) {
                // nextLine[] is an array of values from the line
                data.add(nextLine);
            }
        }
        return data.iterator();
    }).map(line -> RowFactory.create(line));

Reference

• Replace new line (\n) character in csv file - spark scala
• 如何拓展Hadoop的InputFormat为其他分隔符
• opencsv
• Handling multi line data in Spark 2.2.0