来自 Beam PTransform 的 IllegalMutationException
IllegalMutationException from Beam PTransform
这是我写的 Apache Beam PTransform:
public class NormalizeTransform
extends PTransform<PCollection<String>, PCollection<SimpleTable>> {
@Override
public PCollection<SimpleTable> expand(PCollection<String> lines) {
ExtractFields extract_i = new ExtractFields();
PCollection<SimpleTable> table = lines
.apply("Extracting data model fields from lines",
ParDo.of(extract_i));
}
public class ExtractFields extends DoFn<String, SimpleTable> {
@ProcessElement
public void processElement(ProcessContext c){
try {
String line = c.element();
// fill table
for (Table_Struct st: this.struct){
String o = line.substring(st.pos_1, st.pos_2));
this.table.getClass().getField(st.Field_Name).set(
this.table, o);
}
c.output(this.table);
}
}
偶尔会出现以下错误 IllegalMutationException
,这意味着我重复了代码的 运行,有时有效,有时无效。
org.apache.beam.sdk.util.IllegalMutationException: PTransform Transform/Extracting data model fields from lines/ParMultiDo(ExtractFields) mutated value after it was output (new value was ). Values must not be mutated in any way after being output.
at org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.commit(ImmutabilityCheckingBundleFactory.java:135)
at org.apache.beam.runners.direct.EvaluationContext.commitBundles(EvaluationContext.java:214)
at org.apache.beam.runners.direct.EvaluationContext.handleResult(EvaluationContext.java:163)
at org.apache.beam.runners.direct.ExecutorServiceParallelExecutor$TimerIterableCompletionCallback.handleResult(ExecutorServiceParallelExecutor.java:268)
at org.apache.beam.runners.direct.TransformExecutor.finishBundle(TransformExecutor.java:168)
at org.apache.beam.runners.direct.TransformExecutor.run(TransformExecutor.java:109)
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.util.concurrent.FutureTask.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)
我不认为我在我的代码中的任何地方专门更改了值的任何输出。 MutationDetectors 将比较两个值:previousValue 和 newValue。在我的例子中,previousValue 通常是一个输入值,newValue 是另一个输入值。为什么 Transform 会尝试使用一个输入值来修改另一个输入值?
我不确定 this.table
的来源。
但为了帮助您理解错误消息,请记住可能会在多个输入上调用 processElement
。第一次调用将输出 this.table
。下一次调用将在输出之前变异 this.table
。
如果此突变发生在第一次调用输出 this.table
之后且下游代码有机会读取 this.table
之前,您将得到不正确的结果。因此,此错误表明您在输出引用后更改了 this.table
的内容——您不应该这样做。
考虑改为 (1) 输出 this.table
的副本或 (2) 创建 table 作为本地字段。例如:
@ProcessElement
public void processElement(ProcessContext c){
try {
String line = c.element();
Table table = /* create the table */;
// fill table
for (Table_Struct st: this.struct){
String o = line.substring(st.pos_1, st.pos_2));
this.table.getClass().getField(st.Field_Name)
.set(table, o);
}
c.output(table);
}
}
另请注意,在每个 processElement
中执行反射可能比预期的要慢。如果能直接修改字段,可能会更好。
另请注意,强制执行比较反序列化的值,因此它假定反序列化是确定性的。如果对象包含集合,这可能需要注意元素的顺序。
这是我写的 Apache Beam PTransform:
public class NormalizeTransform
extends PTransform<PCollection<String>, PCollection<SimpleTable>> {
@Override
public PCollection<SimpleTable> expand(PCollection<String> lines) {
ExtractFields extract_i = new ExtractFields();
PCollection<SimpleTable> table = lines
.apply("Extracting data model fields from lines",
ParDo.of(extract_i));
}
public class ExtractFields extends DoFn<String, SimpleTable> {
@ProcessElement
public void processElement(ProcessContext c){
try {
String line = c.element();
// fill table
for (Table_Struct st: this.struct){
String o = line.substring(st.pos_1, st.pos_2));
this.table.getClass().getField(st.Field_Name).set(
this.table, o);
}
c.output(this.table);
}
}
偶尔会出现以下错误 IllegalMutationException
,这意味着我重复了代码的 运行,有时有效,有时无效。
org.apache.beam.sdk.util.IllegalMutationException: PTransform Transform/Extracting data model fields from lines/ParMultiDo(ExtractFields) mutated value after it was output (new value was ). Values must not be mutated in any way after being output.
at org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.commit(ImmutabilityCheckingBundleFactory.java:135)
at org.apache.beam.runners.direct.EvaluationContext.commitBundles(EvaluationContext.java:214)
at org.apache.beam.runners.direct.EvaluationContext.handleResult(EvaluationContext.java:163)
at org.apache.beam.runners.direct.ExecutorServiceParallelExecutor$TimerIterableCompletionCallback.handleResult(ExecutorServiceParallelExecutor.java:268)
at org.apache.beam.runners.direct.TransformExecutor.finishBundle(TransformExecutor.java:168)
at org.apache.beam.runners.direct.TransformExecutor.run(TransformExecutor.java:109)
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.util.concurrent.FutureTask.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)
我不认为我在我的代码中的任何地方专门更改了值的任何输出。 MutationDetectors 将比较两个值:previousValue 和 newValue。在我的例子中,previousValue 通常是一个输入值,newValue 是另一个输入值。为什么 Transform 会尝试使用一个输入值来修改另一个输入值?
我不确定 this.table
的来源。
但为了帮助您理解错误消息,请记住可能会在多个输入上调用 processElement
。第一次调用将输出 this.table
。下一次调用将在输出之前变异 this.table
。
如果此突变发生在第一次调用输出 this.table
之后且下游代码有机会读取 this.table
之前,您将得到不正确的结果。因此,此错误表明您在输出引用后更改了 this.table
的内容——您不应该这样做。
考虑改为 (1) 输出 this.table
的副本或 (2) 创建 table 作为本地字段。例如:
@ProcessElement
public void processElement(ProcessContext c){
try {
String line = c.element();
Table table = /* create the table */;
// fill table
for (Table_Struct st: this.struct){
String o = line.substring(st.pos_1, st.pos_2));
this.table.getClass().getField(st.Field_Name)
.set(table, o);
}
c.output(table);
}
}
另请注意,在每个 processElement
中执行反射可能比预期的要慢。如果能直接修改字段,可能会更好。
另请注意,强制执行比较反序列化的值,因此它假定反序列化是确定性的。如果对象包含集合,这可能需要注意元素的顺序。