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")
         .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": [
      {
         "inputs": ["source"],
         "outputs": ["offset", "body"], 
         "name": "read",
         "description": "Reading the data from the file"  
      },
      {
	      "inputs": ["body"],
          "outputs": ["firstName", "lastName", "age", "city", "state"],
          "name": "parse",
          "description": "parsing the input body field"
      },
      {
          "inputs": ["firstName", "lastName"],
          "outputs": ["name"],
          "name": "concat",
          "description": "concatenating the first name and last name"  
      },
      {
          "inputs": ["name"],
          "outputs": ["uniqueid]",
          "name": "create",
          "description": "creating unique id from the name field"  
      }         
   ]
}

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"],
         "outputs": ["offset", "body"], 
         "name": "read",
         "description": "Reading the data from the file"  
      },
      {
	      "inputs": ["body"],

          "outputs": ["firstName", "lastName", "age", "city", "state"],
          "name": "parse",
          "description": "parsing the input body field"
      },
      {
          "inputs": ["firstName", "lastName"],
          "outputs": ["name"],
          "name": "concat",
          "description": "concatenating the first name and last name"  
      },
      {
          "inputs": ["name"],
          "outputs": ["uniqueid]",
          "name": "create",
          "description": "creating unique id from the name field"  
      }         
   ],
 
   "runs": [
      ....
          list of run ids which resulted in this path
      ....
   ]
}
 
 

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",
            }
    ]
 ...
}