Spark - проблема с сериализацией при разборе файлов с использованием OpenCSV - PullRequest
Новая
       14

Spark - проблема с сериализацией при разборе файлов с использованием OpenCSV

0 голосов
/ 17 декабря 2018

Я использую Spark для обработки CSV-файлов.Недавно я заменил ручной разбор CSV-строк на opencsv.Вот упрощенный код 

public class Main {

    public static void main(String[] args) {

        CSVParser parser = new CSVParserBuilder()
                .withSeparator(';')
                .build();

        SparkConf cfg = new SparkConf()
                .setMaster("local[4]")
                .setAppName("Testapp");
        JavaSparkContext sc = new JavaSparkContext(cfg);

        JavaRDD<String> textFile = sc.textFile("testdata.csv", 1);

        List<String> categories = textFile
                .map(line -> parser.parseLine(line)[10])
                .collect();
        System.out.println(categories);
    }
}

К сожалению, этот код не работает.Он генерирует исключение 

Caused by: java.io.NotSerializableException: com.opencsv.CSVParser
Serialization stack:
    - object not serializable (class: com.opencsv.CSVParser, value: com.opencsv.CSVParser@1290c49)
    - element of array (index: 0)
    - array (class [Ljava.lang.Object;, size 1)
    - field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
    - object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class test.Main, functionalInterfaceMethod=org/apache/spark/api/java/function/Function.call:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeStatic test/Main.lambda$main$49bd2722$1:(Lcom/opencsv/CSVParser;Ljava/lang/String;)Ljava/lang/String;, instantiatedMethodType=(Ljava/lang/String;)Ljava/lang/String;, numCaptured=1])
    - writeReplace data (class: java.lang.invoke.SerializedLambda)
    - object (class test.Main$$Lambda$19/429639728, test.Main$$Lambda$19/429639728@72456279)
    - field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function)
    - object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, <function1>)
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
    ... 12 more

Кажется, Spark пытается сериализовать лямбда-выражение, и каким-то образом выражение лямба сохраняет ссылку на parser, что вызывает вышеупомянутую ошибку.

Вопрос: есть ли способчтобы избежать этого исключения и использовать несериализуемую библиотеку в лямбда-выражении, передаваемом в Spark?Я действительно не хочу реализовывать свой собственный парсер csv.

Ответы [ 2 ]

0 голосов
/ 18 декабря 2018

Я понял, что есть очень простое решение моей проблемы.Любое использование внешней библиотеки, которое вызывает проблему сериализации, может быть заключено в статический метод.Ссылка на parser скрыта методом parse.Этот подход явно не идеальное решение, но работает. 

public class Main {

    private static CSVParser parser = new CSVParserBuilder()
            .withSeparator(';')
            .build();

    public static void main(String[] args) {
        SparkConf cfg = new SparkConf()
                .setMaster("local[4]")
                .setAppName("Testapp");
        JavaSparkContext sc = new JavaSparkContext(cfg);

        JavaRDD<String> textFile = sc.textFile("testdata.csv", 1);

        List<String> categories = textFile
                .map(line -> parse(line)[0])
                .collect();
        System.out.println(categories);
    }

    static String[] parse(String line) throws IOException {
        return parser.parseLine(line);
    }
}
0 голосов
/ 17 декабря 2018

Spark поддерживает CSV-файлы из коробки 

import org.apache.spark.sql.Row;
import org.apache.spark.sql.Dataset;

Dataset<Row> df = spark.read().format("csv")
                      .option("sep", ";")
                      .option("header", "true") //or "false" if no headers
                      .load("filename.csv");

Редактировать (повышенный комментарий к основному ответу)

если вам это действительно нужно, вы можете получить СДР из DataFrame с помощью df.javaRDD() хотя предпочтительно использовать API DataSet / DataFrame (см., например, здесь )

...