TupleTag not found in DoFn

user_13558695 0
Viewed 82

I have a DoFn that is supposed to split input into two separate PCollections. The pipeline builds and runs up until it is time to output in the DoFn, and then I get the following exception:

"java.lang.IllegalArgumentException: Unknown output tag Tag<edu.mayo.mcc.cdh.pipeline.PubsubToAvro$PubsubMessageToArchiveDoFn$2.<init>:219#2587af97b4865538>
at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument(Preconditions.java:216)...

If I declare the TupleTags I'm using in the ParDo, I get that error, but if I declare them outside of the ParDo I get a syntax error saying the OutputReceiver can't find the tags. Below is the apply and the ParDo/DoFn:

 PCollectionTuple results = (messages.apply("Map to Archive", ParDo.of(new PubsubMessageToArchiveDoFn()).withOutputTags(noTag, TupleTagList.of(medaPcollection))));

PCollection<AvroPubsubMessageRecord> medaPcollectionTransformed = results.get(medaPcollection);
PCollection<AvroPubsubMessageRecord> noTagPcollectionTransformed = results.get(noTag);

  static class PubsubMessageToArchiveDoFn extends DoFn<PubsubMessage, AvroPubsubMessageRecord> {

final TupleTag<AvroPubsubMessageRecord> medaPcollection = new TupleTag<AvroPubsubMessageRecord>(){};
final TupleTag<AvroPubsubMessageRecord> noTag = new TupleTag<AvroPubsubMessageRecord>(){};

@ProcessElement
public void processElement(ProcessContext context, MultiOutputReceiver out) {
  String appCode;
  PubsubMessage message = context.element();
  String msgStr = new String(message.getPayload(), StandardCharsets.UTF_8);

  try {
    JSONObject jsonObject = new JSONObject(msgStr);
    LOGGER.info("json: {}", jsonObject);
    appCode = jsonObject.getString("app_code");
    LOGGER.info(appCode);
    if(appCode == "MEDA"){
      LOGGER.info("Made it to MEDA tag");
      out.get(medaPcollection).output(new AvroPubsubMessageRecord(
              message.getPayload(), message.getAttributeMap(), context.timestamp().getMillis()));
    } else {
      LOGGER.info("Made it to default tag");
      out.get(noTag).output(new AvroPubsubMessageRecord(
              message.getPayload(), message.getAttributeMap(), context.timestamp().getMillis()));
    }
  } catch (Exception e) {
    LOGGER.info("Error Processing Message: {}\n{}", msgStr, e);
  }
}

}

apache-beam google-cloud-dataflow java
2 Answers

Can you try without MultiOutputReceiver out parameter in the processElement method ?

Outputs are then returned with context.output with passing element and corresponding TupleTag.

Your example only with context :

  static class PubsubMessageToArchiveDoFn extends DoFn<PubsubMessage, AvroPubsubMessageRecord> {

final TupleTag<AvroPubsubMessageRecord> medaPcollection = new TupleTag<AvroPubsubMessageRecord>(){};
final TupleTag<AvroPubsubMessageRecord> noTag = new TupleTag<AvroPubsubMessageRecord>(){};

@ProcessElement
public void processElement(ProcessContext context) {
  String appCode;
  PubsubMessage message = context.element();
  String msgStr = new String(message.getPayload(), StandardCharsets.UTF_8);

  try {
    JSONObject jsonObject = new JSONObject(msgStr);
    LOGGER.info("json: {}", jsonObject);
    appCode = jsonObject.getString("app_code");
    LOGGER.info(appCode);
    if(appCode == "MEDA"){
      LOGGER.info("Made it to MEDA tag");
      context.output(medaPcollection, new AvroPubsubMessageRecord(
              message.getPayload(), message.getAttributeMap(), context.timestamp().getMillis()));
    } else {
      LOGGER.info("Made it to default tag");
      context.output(noTag, new AvroPubsubMessageRecord(
              message.getPayload(), message.getAttributeMap(), context.timestamp().getMillis()));
    }
  } catch (Exception e) {
    LOGGER.info("Error Processing Message: {}\n{}", msgStr, e);
  }
}

I also show you an example that works for me :

public class WordCountFn extends DoFn<String, Integer> {

     private final TupleTag<Integer> outputTag = new TupleTag<Integer>() {};
     private final TupleTag<Failure> failuresTag = new TupleTag<Failure>() {};
    
     @ProcessElement
     public void processElement(ProcessContext ctx) {
        try {
            // Could throw ArithmeticException.
            final String word = ctx.element();
            ctx.output(1 / word.length());
        } catch (Throwable throwable) {
            final Failure failure = Failure.from("step", ctx.element(), throwable);
            ctx.output(failuresTag, failure);
        }
    }
    
    public TupleTag<Integer> getOutputTag() {
       return outputTag;
    }
    
    public TupleTag<Failure> getFailuresTag() {
       return failuresTag;
    }
}

In my first output (good case), no need to pass the TupleTag ctx.output(1 / word.length());

For my second output (failure case), I pass the Failure tag with the corresponding element.

user_92615580

I was able to get around this by making my ParDo an anonymous function instead of a class. I put the whole function inline and had no problem finding the output tags after I did that. Thanks for the suggestions!

user_135586950
Related