Skip to end of metadata
Go to start of metadata

You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 5 Next »

Checklist

  • User Stories Documented
  • User Stories Reviewed
  • Design Reviewed
  • APIs reviewed
  • Release priorities assigned
  • Test cases reviewed
  • Blog post

Introduction 

There have been multiple requests to support sending records to different outputs depending on some condition. Other similar pipelining solutions have this capability.

Goals

The goal is to support pipeline use cases that require sending some records to one output stage and other records to another output stage. 

Use Cases

A pipeline is processing records that contain a union field. The field is a union of 5 possible records. The pipeline developer wants to insert a stage that will have five different output ports, one for each possible schema in the union.

A pipeline is processing records that contain a nullable 'email' field. Most records contain an email, but some do not. The pipeline developer wants to insert a stage that sends all records with empty emails on to a join stage to populate the email field. 

User Stories 

  • As a pipeline developer, I want to insert a stage that sends records to different output ports
  • As a pipeline developer, I want to know how many output ports a stage supports
  • As a pipeline developer, I want to see the schema for each output port
  • As a plugin developer, I want to specify which records are sent to which output port
  • As a plugin developer, I want to be able to specify different output ports depending on input schema and configuration
  • As a plugin developer, I want to be able to modify records before they are sent to an output port

Design

We will introduce a new 'splitter' plugin type that can send records to multiple outputs. Each output can be configured with its own schema. The plugin interface will be:

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);
}

For example, a plugin that splits based on a union field can be implemented like so:

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;
  }
}

 

The pipeline config will need to be enhanced. When a splitter plugin is involved, connections from a splitter to other stages must also specify which splitter output is involved:

 

{
  "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"}
  ]
}

 

Approach

Approach #1

Approach #2

API changes

New Programmatic APIs

new Splitter plugin type and MultiOutput configurers.

Deprecated Programmatic APIs

None

New REST APIs

None

Deprecated REST API

None

CLI Impact or Changes

None

UI Impact or Changes

UI will need enhancements to support multiple outputs. It will need some way to display and connect multiple outputs of a stage and manage schema for each output.

Security Impact 

None

Impact on Infrastructure Outages 

None

Test Scenarios

Test IDTest DescriptionExpected Results
   
   
   
   

Releases

Release X.Y.Z

Release X.Y.Z

Related Work

  • Work #1
  • Work #2
  • Work #3

 

Future work

  • No labels