PCollectionView<List<Foo>>的访问元素: Google Cloud Dataflow/Apache Beam

import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TypeDescriptors;

public class CreateViewFn extends DoFn<Foo, KV<Integer, Foo>> {
    @ProcessElement
    public void processElement(ProcessContext c) {
        Foo foo = c.element();
        c.output(KV.of(foo.getId(), foo));
    }
}

PCollection<Foo> foos = ...; // 你的Foo类型的PCollection
PCollectionView<Map<Integer, Foo>> fooView =
    foos.apply(ParDo.of(new CreateViewFn()))
        .apply(View.<Integer, Foo>asMap());

import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;

public class AccessViewFn extends DoFn<Integer, Foo> {
    private final PCollectionView<Map<Integer, Foo>> fooView;

    public AccessViewFn(PCollectionView<Map<Integer, Foo>> fooView) {
        this.fooView = fooView;
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        Integer key = c.element();
        Map<Integer, Foo> fooMap = c.sideInput(fooView);
        Foo foo = fooMap.get(key);
        if (foo != null) {
            c.output(foo);
        }
    }
}

PCollection<Integer> keys = ...; // 你的键的PCollection
PCollection<Foo> accessedFoos =
    keys.apply(ParDo.of(new AccessViewFn(fooView)).withSideInputs(fooView));