Apache Beam：不可见参数类型异常

Question

我在 Apache Beam 中构建了一个小函数来执行 lookup/join：给定一个映射 A 到 B 的集合，另一个映射 B 到 C 的集合，return 一个映射 A 到 C 的集合.

class Main {
    private static <A,B,C> PCollection<KV<A,C>> lookup(
            PCollection<KV<A,B>> collection,
            PCollection<KV<B,C>> lookup
    ){
        var leftTag = new TupleTag<A>();
        var rightTag = new TupleTag<C>();

        return KeyedPCollectionTuple.of(leftTag, collection.apply(KvSwap.create()))
                .and(rightTag, lookup)
                .apply(CoGroupByKey.create())
                .apply(ParDo.of(new DoFn<KV<B, CoGbkResult>, KV<A, C>>() {
                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        var value = c.element().getValue();
                        var right = value.getOnly(rightTag);

                        for (var i : value.getAll(leftTag)) {
                            c.output(KV.of(i, right));
                        }
                    }
                }));
    }

    public static void main(String[] args) {

        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline p = Pipeline.create(options);

        PCollection<KV<String, Long>> test = p.apply(Create.of(KV.of("a", 1L), KV.of("b", 2L)))
                .setTypeDescriptor(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.longs()));

        PCollection<KV<Long, String>> test2 = p.apply(Create.of(KV.of(1L, "a"), KV.of(2L, "b")))
                .setTypeDescriptor(TypeDescriptors.kvs(TypeDescriptors.longs(), TypeDescriptors.strings()));

        var c = lookup(test, test2)
                .setTypeDescriptor(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.strings()));

        p.run().waitUntilFinish();
    }
}

不幸的是，当运行管道时出现以下错误：

Exception in thread "main" org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.util.concurrent.UncheckedExecutionException: java.lang.IllegalStateException: Invisible parameter type of Main arg0 for public Main$DoFnInvoker(Main)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2214)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.cache.LocalCache.get(LocalCache.java:4053)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4057)
    ...
Caused by: java.lang.IllegalStateException: Invisible parameter type of Main arg0 for public Main$DoFnInvoker(Main)
    at org.apache.beam.repackaged.beam_sdks_java_core.net.bytebuddy.dynamic.scaffold.InstrumentedType$Default.validated(InstrumentedType.java:925)
    at org.apache.beam.repackaged.beam_sdks_java_core.net.bytebuddy.dynamic.scaffold.MethodRegistry$Default.prepare(MethodRegistry.java:465)
    at org.apache.beam.repackaged.beam_sdks_java_core.net.bytebuddy.dynamic.scaffold.subclass.SubclassDynamicTypeBuilder.make(SubclassDynamicTypeBuilder.java:170)
    ...

我似乎找不到很多关于此异常含义的文档。我认为这与类型擦除有关，并尝试用 TypeDescriptors 随意地填充代码，但没有成功。我该怎么做才能修复此错误？

Answer 1

原来这是由于使用 JDK 10 引起的，在编写时 Beam 不支持它。切换到 JDK 8 解决了这个问题。

Apache Beam：不可见参数类型异常

Apache Beam: Invisible parameter type exception

apache-beam