我正在使用 Spark 处理 csv 文件。最近我用 opencsv 替换了手动 CSV 行解析。这是简化的代码
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 试图序列化 lambda 表达式,而 lamba 表达式以某种方式保持对 parser
的引用,这导致了上述错误。
问题是:是否有任何方法可以避免该异常并在传递给 Spark 的 lambda 表达式中使用不可序列化的库?我真的不想实现自己的 csv 解析器。
最佳答案
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");
编辑(对主要答案的推荐评论)
如果你真的需要它,你可以使用 df.javaRDD()
从 DataFrame 中获取 RDD
尽管最好使用 DataSet/DataFrame API(例如参见 here)
关于java - Spark - 使用 OpenCSV 解析文件的序列化问题,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/53817643/