断言错误:断言失败: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();

}

推荐答案

我在创建扩展 org.apache.spark.util.AccumulatorV2 的自定义累加器时遇到了类似的问题.原因是 override def isZero: Boolean 方法中的逻辑不正确.所以基本上当你默认调用 copyAndReset 方法时，它会调用 copy() 然后 reset() 你的 isZero()应该返回true.如果您查看 AccumulatorV2 源代码，则检查的位置是:

I was facing similar issue when I created custom accumulator that extends org.apache.spark.util.AccumulatorV2. The cause was improper logic in override def isZero: Boolean method. So basically when you copyAndReset method called by default it calls copy() then reset() your isZero() should return true. If you look at the AccumulatorV2 source that is where is a check is:

// Called by Java when serializing an object
final protected def writeReplace(): Any = {
 if (atDriverSide) {
   if (!isRegistered) {
     throw new UnsupportedOperationException(
       "Accumulator must be registered before send to executor")
   }
   val copyAcc = copyAndReset()
   assert(copyAcc.isZero, "copyAndReset must return a zero value copy")
   copyAcc.metadata = metadata
   copyAcc
 } else {
   this
 }
}

特别是这部分

 val copyAcc = copyAndReset()
 assert(copyAcc.isZero, "copyAndReset must return a zero value copy")

希望有帮助.快乐火花！

Hope it helps. Happy sparking!

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