Spark, NegativeArraySizeException, когда sequencefile - PullRequest
Новая
       7

Spark, NegativeArraySizeException, когда sequencefile

1 голос
/ 03 апреля 2019

Используемая мной искра - 2,3.

У меня есть этот фрагмент кода, который читает файлы последовательности в «hdfspath» (по этому пути около 20 файлов, и каждый файл занимает около 60 МБ), 

SparkSession spark = ...;
JavaSparkContext jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());
JavaPairRDD<BytesWritable, BytesWritable> temp = jsc.sequenceFile(hdfspath, BytesWritable.class, BytesWritable.class);
temp.take(1);

И это дает мне эту ошибку, 

19/04/03 14:50:18 INFO CodecPool: Got brand-new decompressor [.gz]
19/04/03 14:50:18 INFO CodecPool: Got brand-new decompressor [.gz]
19/04/03 14:50:18 INFO CodecPool: Got brand-new decompressor [.gz]
19/04/03 14:50:18 INFO CodecPool: Got brand-new decompressor [.gz]
19/04/03 14:50:18 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.NegativeArraySizeException
    at org.apache.hadoop.io.BytesWritable.setCapacity(BytesWritable.java:144)
    at org.apache.hadoop.io.BytesWritable.setSize(BytesWritable.java:123)
    at org.apache.hadoop.io.BytesWritable.readFields(BytesWritable.java:179)
    at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:71)
    at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:42)
    at org.apache.hadoop.io.SequenceFile$Reader.deserializeKey(SequenceFile.java:2606)
    at org.apache.hadoop.io.SequenceFile$Reader.next(SequenceFile.java:2597)
    at org.apache.hadoop.mapred.SequenceFileRecordReader.next(SequenceFileRecordReader.java:82)
    at org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:277)

hdfs-файлы, которые я пытаюсь прочитать, являются выходом старого задания mapreduce с настройкой вывода, подобной этой, 

job.setOutputKeyClass(BytesWritable.class);
job.setOutputValueClass(BytesWritable.class);
job.setOutputFormatClass(SequenceFileAsBinaryOutputFormat.class);
SequenceFileAsBinaryOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

Я изучил метод org.apache.hadoop.io.BytesWritable.setCapacity (...), 

public void setSize(int size) {
if (size > getCapacity()) {
  setCapacity(size * 3 / 2);
}
this.size = size;
}

Каким-то образом параметр размера равен 808464432 и вызывает переполнение при выполнении размера * 3, что в конечном итоге вызывает исключение NegativeArraySizeException.

Может кто-нибудь помочь объяснить, как это происходит и как это исправить?

1 Ответ

0 голосов
/ 08 апреля 2019

разобрался. Используйте JavaSparkContext#newAPIHadoopFile вместо JavaSparkContext#sequenceFile.

...