File------->Parser------->Name-Concatenator------->IdGenerator------->Store
 
 

// Identifies the source from where the data is read
public class Source {
   String namespace;
   String name; // could be the name of the dataset, Kafka topic etc.			
   String description; // Optional description associated with the source
}


// Represents the input for the operation
public interface Input {
  // Name of the input
  String getName(); 
 
  // Optional description associated with the input
  String getDescription();
 
  // Get the type associated with the input
  // Probably can be used to distinguish between different types of inputs such as source input or field input.
  Type getType(); 
}
 
// Represent Source as an input for the Field operation
public class SourceInput extends AbstractInput {
   Source source;
}
 
// Represent Field as an input for the Field operation
public class FieldInput extends AbstractInput {
   Schema.Field field;   
}
 
public class FieldOperation {
   // Operation name
   String name;
 
   // Optional detailed description about the operation
   String description;
 
   // Set of input fields participate in the operation	
   Set<Input> inputs;
 
   // Set of output fields generated as a part of this operation
   // Question: Can this be null? For example in case of Drop operation. 
   // However if the field is dropped and its not present in the destination dataset can it be reached?
   Set<Schema.Field> outputs;
 
   // Builder for the FieldOperation
   public static Builder {
      String name;
      String description;
      Set<Input> inputs;
      Set<Schema.Field> outputs;
 
      private Builder() {
        inputs = new HashSet<>();
        outputs = new HashSet<>();
      }
 
      public Builder setName(String name) {
         this.name = name;   
         return this;
      }
 
      public Builder setDescription(String description) {
         this.description = description;
         return this;
      }
 
      public Builder addInputSource(Source source) {
         Input input = new SourceInput(source);
         inputs.add(input);
         return this;
      }
 
      public Builder addInputField(Field field) {
         Input input = new FieldInput(field);
         inputs.add(input);
         return this;
      }
 
      public Builder addOutput(Field field) {
         outputs.add(field);
         return this; 
      }    
   }		
}

/**
 * This interface provides methods that will allow programs to record the field level operations.
 */
public interface LineageRecorder {
    /**
 	 * Record the field level operations against the given destination.
 	 *
 	 * @param destination the destination for which to record field operations
 	 * @param fieldOperations The list of field operations.
 	*/
    void record(Destination destination, List<FieldOperation> fieldOperations);
}
 
public class Destination {
   String namespace;
   String id;
   String description;
}

public class ETLMapReduce extends AbstractMapReduce {
...
   public void initialize() throws Exception {
      MapReduceContext context = getContext();
	  ...
	  List<FieldOperation> operations = new ArrayList<>();
      Source source = Source.from("myns", "user_data", "files contains user information");
 
      FieldOperation.Builder builder = new FieldOperation.Builder();
      builder
         .setName("read")
         .setDescription("Reading the data from the file")
         .addInputSource(source)
         .addOutput(offsetField)
         .addOutput(bodyField);      
      operations.add(builder.build());
       
      builder = new FieldOperation.Builder();
      builder
         .setName("parse")
         .setDescription("parsing the input body field")
         .addInput(bodyField)
         .addOutput(firstNameField)
         .addOutput(lastNameField)
         .addOutput(ageField)
         .addOutput(cityField)
         .addOutput(stateField);
      operations.add(builder.build());


      builder = new FieldOperation.Builder();
      builder
         .setName("concat")
         .setDescription("concatenating the first name and last name")
         .addInput(firstNameField)
         .addInput(lastNameField)
         .addOutput(nameField);
      operations.add(builder.build());


      builder = new FieldOperation.Builder();
      builder
         .setName("drop")
         .setDescription("deleting the first name field")
         .addInput(firstNameField)
      operations.add(builder.build());
 
      builder = new FieldOperation.Builder();
      builder
         .setName("drop")
         .setDescription("deleting the last name field")
         .addInput(lastNameField)
      operations.add(builder.build());

      builder = new FieldOperation.Builder();
      builder
         .setName("create")
         .setDescription("creating unique id from the name field")
         .addInput(nameField)
         .addOutput(idField)
      operations.add(builder.build());

      context.record(Destination.from("myns", "mytableds"), operations);
      ...    
   }
...
}

public interface StageContext {
...
    /**
 	 * Record the field level mutations.
 	 *
 	 * @param fieldMutations The list of field mutations
 	 /
	void recordFieldMutations(List<FieldMutation> fieldMutations); 
...
}

{
 ...
    "stages" : [
		{
                "name": "Projection",
                "plugin": {
                    "name": "Projection",
                    "type": "transform",
                    "label": "Projection",
                    "artifact": {
                        "name": "core-plugins",
                        "version": "1.8.5-SNAPSHOT",
                        "scope": "SYSTEM"
                    },
                    "properties": {
                        "drop": "headers",
                        "rename": "ts:seqId,body:f1"
                    }
                },
 
                "fieldMutations": [
                    {
						"inputFields": [
							{\"name\":\"ts\",\"type\":\"long\"}
						],
						"mutation": "rename",
                        "description": "renaming timestamp to the sequence id",
						"outputField": {\"name\":\"seqId\",\"type\":\"long\"}						
					},
                    {
						"inputFields": [
							{\"name\":\"body\",\"type\":\"string\"}
						],
						"mutation": "rename",
                        "description": "renaming body to the f1",
						"outputField": {\"name\":\"f1\",\"type\":\"string\"}
					},
                    {
						"inputFields": [
							{\"name\":\"headers\",\"type\":{\"type\":\"map\",\"keys\":\"string\",\"values\":\"string\"}
						],
						"mutation": "delete",
                        "description": "deleting headers"
					}
                ],
                "outputSchema": "{\"type\":\"record\",\"name\":\"etlSchemaBody\",\"fields\":[{\"name\":\"seqId\",\"type\":\"long\"},{\"name\":\"f1\",\"type\":\"string\"}]}",
                "inputSchema": [
                    {
                        "name": "Stream",
                        "schema": "{\"name\":\"etlSchemaBody\",\"type\":\"record\",\"fields\":[{\"name\":\"ts\",\"type\":\"long\"},{\"name\":\"headers\",\"type\":{\"type\":\"map\",\"keys\":\"string\",\"values\":\"string\"}},{\"name\":\"body\",\"type\":\"string\"}]}"
                    }
                ],
                "type": "transform",
                "label": "Projection",
                "icon": "icon-projection",
            }
    ]
 ...
}