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);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")
         .addInputField(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")
         .addInputField(firstNameField)
         .addInputField(lastNameField)
         .addOutput(nameField);
      operations.add(builder.build());


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

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

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

{
   "destination": {
      "namespace": "lineage",
      "name": "KafkaSink",
      "description": "Kafka broker running at address localhost:9092"  
   },
   "source": {
      "namespace": "myns",
      "name": "user_data",
      "description": "files contains user information"
   },
   "operations": [
      {
          return this;"inputs": ["source"],
       }   "outputs": ["offset", "body"], 
  public Builder addOutput(Field field) {   "name": "read",
     outputs.add(field);    "description": "Reading the data from the returnfile" this; 
      },
      {
}		 } 

GET /v3/namespaces/<namespace-id>/datasets/<dataset-id>/fields/<field-name>/lineage?start=<start-ts>&end=<end-ts>
 
Where:
namespace-id: namespace name
dataset-id: dataset name
field-name: name of the field for which lineage information to be retrieved
start-ts: starting timestamp(inclusive) in seconds
end-ts: ending timestamp(exclusive) in seconds for lineage

{
  "fieldName": "uniqueId",
  "startTimeInSeconds": 1442863938,
  "endTimeInSeconds": 1442881938,
  "paths": [
   ....
       list of paths which represent the different ways field is created  
   ....
  ]  
}
 
Each path will look as follows:
 
{
   "operations": [
      {
         "inputs": ["source"],
         builder = new FieldOperation.Builder();"outputs": ["offset", "body"], 
      builder   "name": "read",
     .setName("concat")    "description": "Reading the data from the .setDescription(file"concatenating  the
first name and last name")  },
      {
.addInputField(firstNameField)	      "inputs": ["body"],

 .addInputField(lastNameField)          .addOutput(nameField);
      operations.add(builder.build());"outputs": ["firstName", "lastName", "age", "city", "state"],
          builder = new FieldOperation.Builder();"name": "parse",
        builder  "description": "parsing the input body field"
  .setName("drop")    },
     .setDescription("deleting the{
first name field")        "inputs":  .addInputField(firstNameField)["firstName", "lastName"],
      operations.add(builder.build());     "outputs": ["name"],
  builder = new FieldOperation.Builder();     "name": "concat",
builder          .setName("drop")
     "description": "concatenating the first name and last name"  
 .setDescription("deleting the last name field") },
      {
 .addInputField(lastNameField)       operations.add(builder.build());  "inputs": ["name"],
    builder = new FieldOperation.Builder();   "outputs": ["uniqueid]",
  builder          .setName("name": "create"),
          .setDescription("description": "creating unique id from the name field")  
      }         
.addInputField(nameField)   ],
 
    .addOutput(idField)"runs": [
      operations...add(builder.build());
       context.record(Destination.from("myns", "mytableds"), operations); list of run ids which resulted ...in this path
     } ....
   ]
}