Apache Beam returns "Input values must not be mutated in any way." 使用本地直接运行程序时

Question

我写了一个 Apache Beam DoFn

static class FillLocation extends DoFn<TrackingRequest, TrackingRequest> {
        @ProcessElement
        public void processElement(ProcessContext c) {    
            TrackingRequest rq = c.element();
            rq.location = getLocationFromIP(rq.IP);         
            c.output(rq);
        }
}

在本地测试时它给了我这个错误 PTransform .. class.....

的非法变异值 ..

 Input values must not be mutated in any way.
    at org.apache.beam.runners.direct.ImmutabilityEnforcementFactory$ImmutabilityCheckingEnforcement.verifyUnmodified(ImmutabilityEnforcementFactory.java:96)
    at org.apache.beam.runners.direct.ImmutabilityEnforcementFactory$ImmutabilityCheckingEnforcement.afterElement(ImmutabilityEnforcementFactory.java:71)
    at org.apache.beam.runners.direct.TransformExecutor.processElements(TransformExecutor.java:149)
    at org.apache.beam.runners.direct.TransformExecutor.run(TransformExecutor.java:107)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

您的函数修改了输入 TrackingRequest 元素的位置字段。数据流不允许这样做。

doc 说：

The current element of the input PCollection is returned by c.element(). It should be considered immutable. The Dataflow runtime will not mutate the element, so it is safe to cache, etc. The element should not be mutated by any of the DoFn methods, because it may be cached elsewhere, retained by the Dataflow runtime, or used in other unspecified ways.

您可以创建输入元素的副本，修改字段，然后将副本作为输出发送。