AssertionError:断言失败:copyAndReset必须返回零值副本 [英] AssertionError: assertion failed: copyAndReset must return a zero value copy

问题描述

当我将ParDo.of(new ParDoFn())应用于名为textInput的PCollection时，程序将引发此异常.但是，当我删除.apply(ParDo.of(new ParDoFn()))时，程序是正常的.

When I applied ParDo.of(new ParDoFn()) to PCollection named textInput, The program throws this Exception. But The Program is normal when I delete .apply(ParDo.of(new ParDoFn())).

//SparkRunner

//SparkRunner

private static void testHadoop(Pipeline pipeline){
    Class<? extends FileInputFormat<LongWritable, Text>> inputFormatClass =
            (Class<? extends FileInputFormat<LongWritable, Text>>)
                    (Class<?>) TextInputFormat.class;
    @SuppressWarnings("unchecked")  //hdfs://localhost:9000
            HadoopIO.Read.Bound<LongWritable, Text> readPTransfom_1 = HadoopIO.Read.from("hdfs://localhost:9000/tmp/kinglear.txt",
            inputFormatClass,
            LongWritable.class,
            Text.class);
    PCollection<KV<LongWritable, Text>> textInput = pipeline.apply(readPTransfom_1)
            .setCoder(KvCoder.of(WritableCoder.of(LongWritable.class), WritableCoder.of(Text.class)));

    //OutputFormat
    @SuppressWarnings("unchecked")
    Class<? extends FileOutputFormat<LongWritable, Text>> outputFormatClass =
            (Class<? extends FileOutputFormat<LongWritable, Text>>)
                    (Class<?>) TemplatedTextOutputFormat.class;

    @SuppressWarnings("unchecked")
    HadoopIO.Write.Bound<LongWritable, Text> writePTransform = HadoopIO.Write.to("hdfs://localhost:9000/tmp/output", outputFormatClass, LongWritable.class, Text.class);

    textInput.apply(ParDo.of(new ParDoFn())).apply(writePTransform.withoutSharding());

    pipeline.run().waitUntilFinish();

}

推荐答案

您在哪个Spark版本上运行?根据我的经验，Spark 2.x AccumulatorV2引发了您遇到的错误，Spark运行程序当前支持Spark 1.6.

Which Spark version are you running on top ? From my experience the error you're getting is thrown by Spark 2.x AccumulatorV2, Spark runner currently supports Spark 1.6.

这篇关于AssertionError:断言失败:copyAndReset必须返回零值副本的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！