问题描述
限时送ChatGPT账号..当我将 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屋!
更多推荐
[db:关键词]
发布评论