{
  "stages": [
    { "name": "A", ... },
    { "name": "splitter", ... },
    { "name": "B", ... },
    { "name": "C", ... }
  ],
  "connections": [
    { "from": "A", "to": "splitter" },
    { "from": "splitter", "output": "0", "to": "B"},
    { "from": "splitter", "output": "1", "to": "C"}
  ]
}

public abstract class Splitter<T> implements StageLifecycle<SplitterContext>, MultiOutputPipelineConfigurable {
  public static final String PLUGIN_TYPE = "splitter";
 
  /**
   * Configure the pipeline. This is run once when the pipeline is being published.
   * This is where you perform any static logic, like creating required datasets, performing schema validation,
   * setting output schema, and things of that nature.
   *
   * @param configurer the configurer used to add required datasets and streams
   */
  @Override
  public void configurePipeline(MultiOutputPipelineConfigurer configurer) {
    // no-op
  }
 
  /**
   * Initialize the splitter.
   */
  @Override
  public void initialize(SplitterContext context) throws Exception {
    //no-op
  }

  /**
   * Destroy the splitter.
   */
  @Override
  public void destroy() {
    //no-op
  }
 
  public abstract void transform(T input, MultiOutputEmitter<T> emitter);
}
 

public interface MultiOutputEmitter<T> {
 
  public emit(String output, Object record);
 
  public emitError(InvalidEntry<T> invalidEntry);
}
 
public interface MultiOutputPipelineConfigurable {
 
  void configurePipeline(MultiOutputPipelineConfigurer multiOutputPipelineConfigurer);
}
 
public interface MultiOutputPipelineConfigurer extends PipelineConfigurer {
  
  MultiOutputStageConfigurer getMultiOutputStageConfigurer();
}
 
public interface MultiOutputStageConfigurer {
 
  Schema getInputSchema();
 
  void setOutputSchemas(Map<String, Schema> outputSchemas);
}

public class UnionFieldSplitter extends Splitter<StructuredRecord> {
  private Conf conf;
 
  public void configurePipeline(MultiOutputPipelineConfigurer configurer) {
    MultiOutputStageConfigurer stageConfigurer = configurer.getMultiOutputStageConfigurer();
    Schema inputSchema = stageConfigurer.getInputSchema();
    if (inputSchema != null) {
      stageConfigurer.setOutputs(getOutputs(inputSchema));
    }
  }
 
  @Override
  public void transform(StructuredRecord input, MultiOutputEmitter<StructuredRecord> emitter) {
    Schema.Field splitField = input.get(conf.field);
    if (splitField == null) {
      emitter.emitError(new InvalidEntry<>(100, "Field " + conf.field + " does not exist in input schema.", input));
    }
    if (splitField.getType() != Schema.Type.UNION) {
      emitter.emitError(new InvalidEntry<>(101, "Field " + conf.field + " is not of type union, but is of type " + splitField.getType()));
    }
    
    StructuredRecord val = input.get(conf.field);
    List<Schema.Field> fields = new ArrayList<>(input.getSchema().getFields().size());
    for (Schema.Field field : input.getSchema().getFields()) {
      fields.add(field.getName().equals(conf.field) ? val.getSchema() : field);
 
    }
 
    int i = 0;
    for (Schema unionSchema : splitField.getSchema().getUnionSchemas()) {
      if (val.getSchema().equals(unionSchema)) {
        break;
      }      
      i++;
    }
    StructuredRecord.Builder builder = StructuredRecord.builder(Schema.recordOf(input.getSchema().getName() + "." + i);
    for (Schema.Field field : input.getSchema().getFields()) {
      builder.set(field.getName(), input.get(field.getName());
    }
    emitter.emit(String.valueOf(i), builder.build());
  }
 
  private Map<String, Schema> getOutputs(Schema inputSchema) {
    Schema.Field splitField = inputSchema.getField(conf.field);
    if (splitField == null) {
      throw new IllegalArgumentException("Field " + conf.field + " does not exist in input schema.");
    }
    if (splitField.getType() != Schema.Type.UNION) {
      throw new IllegalArgumentException("Field " + conf.field + " must be a union field, but is of type " + splitField.getType());
    }

    List<Schema.Field> fields = new ArrayList<>();
    int i = 0;
    int splitFieldIndex = 0;
    for (Schema.Field inputField : inputSchema.getFields()) {
      if (!inputField.getName().equals(conf.field)) {
        fields.set(i, inputField);
      } else {
        splitFieldIndex = i;
      }
      i++;
    }

    Map<String, Schema> outputs = new HashMap<>();
    i = 0;
    for (Schema unionSchema : splitField.getSchema().getUnionSchemas()) {
      fields.set(splitFieldIndex, unionSchema);
      outputs.put(String.valueOf(i), Schema.recordOf(inputSchema.getName() + "." + i, fields));
      i++;
    }
  }
 
  @Path("outputSchemas")
  public Map<String, Schema> outputSchemas(GetSchemasRequest request) {
    this.conf = request;
    return getOutputs(request.inputSchema);
  }
 
  public static class Conf extends PluginConfig {
    @Description("The field to split on. Must be a union field.")
    private String field;
  }
 
  public static class GetSchemasRequest extends Conf {
    private Schema inputSchema;
  }
}