SKHolmes opened a new issue, #27753:
URL: https://github.com/apache/beam/issues/27753
### What happened?
First of all let me pre-face this with, I am still new to Apache Beam so
apologies if I made a simple mistake or am misunderstanding something.
We are using Beam to create a generic data streaming pipeline that reads
from a messaging service and process the data depending on configurations
present in the message. I believe this means at some point the Transformations
will need to return a PCollectionTuple to break the data into their relevant
PCollections depending on the different data types. However when attempting to
access the returned PCollections using their tuple-tags in the app only the
default PCollection/TupleTag is present.
> Exception in thread "main" java.lang.IllegalArgumentException: TupleTag
Tag<com.example.transforms.MessageParser.lambda$create$0:37#bb20b45fd4d95138>
not found in this PCollectionTuple tuple
**Main App**
```
var pipeline = Pipeline.create(options);
MessageParser messageParser = MessageParser.create();
Map<PipelineType, TupleTag<Message>> pipelineTypeMap =
messageParser.getPipelineTypeMap();
PCollectionTuple pCollectionTuple = pipeline.apply("Reading Pub/Sub",
PubsubIO.readMessages().fromSubscription(options.getPubSubSubscription()))
.apply(Window.<PubsubMessage>into(new
GlobalWindows())
.triggering(AfterPane.elementCountAtLeast(1))
.discardingFiredPanes()
)
.apply("Parsing Pub/Sub Message",
messageParser);
// Fails on line below
PCollection<Message> curvePCollection =
pCollectionTuple.get(pipelineTypeMap.get(PipelineType.CURVE_INGESTION));
PCollection<Message> testPCollection =
pCollectionTuple.get(pipelineTypeMap.get(PipelineType.TEST));
pipeline.run();
```
**MessageParser**
```
public class MessageParser extends PTransform<PCollection<PubsubMessage>,
PCollectionTuple> {
private TupleTagList tags;
private TupleTag<Message> defaultTupleTag;
private Map<PipelineType, TupleTag<Message>> pipelineTypeMap;
private MessageParser(TupleTag<Message> defaultTupleTag, TupleTagList
tags, Map<PipelineType, TupleTag<Message>> pipelineTypeMap) {
this.defaultTupleTag = defaultTupleTag;
this.tags = tags;
this.pipelineTypeMap = pipelineTypeMap;
}
public static MessageParser create() {
Map<PipelineType, TupleTag<Message>> pipelineTypeMap = new
HashMap<>();
TupleTagList tags = TupleTagList.empty();
EnumSet.allOf(PipelineType.class)
.forEach(pipelineType -> {
TupleTag<Message> t = new TupleTag<>();
pipelineTypeMap.put(pipelineType, t);
tags.and(t);
});
TupleTag<Message> defaultTag = new TupleTag<>();
// Added for visibility
pipelineTypeMap.put(null, defaultTag);
return new MessageParser(defaultTag, tags, pipelineTypeMap);
}
@Override
public PCollectionTuple expand(PCollection<PubsubMessage> input) {
return input.apply("Running read: ", ParDo.of(new
MessageParserFn(pipelineTypeMap)).withOutputTags(defaultTupleTag, tags));
}
public Map<PipelineType, TupleTag<Message>> getPipelineTypeMap() {
return pipelineTypeMap;
}
}
```
**PipelineType**
```
public enum PipelineType {
CURVE_INGESTION,
TEST
}
```
I suspect it's failing because I am attempting to interact with the
PCollectionTuple outside of an execution of the pipeline transforms. But
reading into the documentation here,
[https://beam.apache.org/documentation/pipelines/design-your-pipeline/#a-single-transform-that-produces-multiple-outputs](https://beam.apache.org/documentation/pipelines/design-your-pipeline/#a-single-transform-that-produces-multiple-outputs)
the provided example seems to do exactly what I want to do and gets the
PCollections post tagged output. What am I misunderstanding about
multi-outputs? Is there a bug? Is the documentation wrong/unclear?
I can provide more source if required but I am hoping it will be something
obvious I am doing wrong
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
