Introduction 

CDAP currently captures lineage at the dataset level. With lineage, users can tell the program that read from or wrote to a dataset. It can help users determine which program wrote to/read from a dataset in a given timeframe.

However, as a platform, CDAP understands schemas for most datasets. Schemas contain fields. It would be useful to be able to drill into how a field in a particular dataset was used (CREATE/READ/WRITE/DELETE) in a given time period.

Goals

• Provide CDAP platform support (in the form of API and storage) to track field level lineage.
• Pipelines can then expose this functionality to the plugins.
• Plugins (such as wrangler) will need to be updated to use this feature.

Use Cases 

User Stories

1. Spark program can perform various transformations on the input fields of the dataset to generate new fields. For example concatenate the first_name and last_name fields of the input dataset, so that resultant dataset only has Name as field. As a developer of CDAP program(for example CDAP Spark program), I should be able to provide these transformations so that I will know later about how the field Name was generated.
   
2. Similar transformations on the fields can be done in the CDAP plugins as well. Plugin developer should be able to provide such transformations.
3. Few plugins such as Javascript transform, Python transform etc execute the custom code provided by the pipeline developer. Pipeline developer in this case should be able to provide the field transformations through the plugin config UI.

Design

Consider a following pipeline: 

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

Consider a following sample structured record which was processed by the pipeline along with the operations that happened on the fields:

There are two ways to show the lineage graph for the field id as -

1. Simple view: In this view, the lineage graph for each field has only 2 types of nodes - one belonging to the source datasets from which the field is ultimately created and the field itself. The edge between these nodes represents the set of operations that have been performed on the source nodes. Any intermediate generated fields will not be shown.
   
   

   Code Block
   parse concat create body--------------->id

2. Detailed view: In this view, each node of the graph represents the field (including any intermediate field generated) and each edge represents the single operation which transforms the input node into the corresponding output node.

   Code Block
   |---first_name---->| (parse)| | (concat) (create) body------>| |-------->name---------->id | | |---last_name----->|

CDAP Platform API changes

We will add FieldOperation class in the cdap-api to represent individual field operation.

// Identifies the source from where the data is read
public class FieldOperationSource {
   //String Operationnamespace;
name    String name;  // could be the //name Optionalof detailedthe descriptiondataset, aboutKafka the operationtopic etc.			
   String description;
 
 // Optional description associated with the source
}


// SetRepresents ofthe input fields participate infor the operation	
public interface  Set<FieldPath> inputFields;
 Input {
   // SetName of outputthe fieldsinput
generated as a part of this operationString getName(); 
 
   // Question:Optional Candescription thisassociated bewith null?the Forinput
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<FieldPath> outputFields;		
}
 
public class FieldPath {
   // Represents the path to the field in the Schema. 
   // We want the field to be uniquely determined. 
   // So if the field with the same name exists at different level of nesting 
   // in the schema we will accept the complete ordered path from root to that field.
   List<String> pathString 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 Originaldetailed sourcedescription informationabout fromthe whereoperation
the field is coming.String description;
 
 Source source; // Set of }input  fields publicparticipate classin Sourcethe {operation	
   StringSet<Input> namespaceinputs;
  
 String id; // couldSet beof theoutput namefields ofgenerated theas dataset,a Kafkapart topic etc.			
 of this operation
 String description; // OptionalQuestion: descriptionCan associatedthis withbe thenull? source
}

Once constructed following platform api can be used to record these field operations:

/**
 * This interface provides methods that will allow programs to recordFor example in case of Drop operation. 
   // However if the field levelis operations.dropped and */its publicnot interfacepresent LineageRecorderin {the destination dataset can it /**be reached?
	 * 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;
}

LineageRecorder will be available in the initialize method of the CDAP programs through program runtime context.

Sample program code to record the operations as done for the above pipeline from ETLMapReduce. Note that this code is called from the CDAP program and not from the plugin.

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");
      FieldPath inputFieldPath = new FieldPath(null /* No input fields for source */, source);
      FieldPath outputFieldPath = new FieldPath(Arrays.asList("body"), source);
      
      Set<FieldPath> input = new HashSet<>();
      input.add(inputFieldPath);
      Set<FieldPath> output = new HashSet<>();
      output.add(outputFieldPath);
 
      FieldOperation operation = new FieldOperation("read", "reading the information about users from source", input, output);
      operations.add(operation);
 
      input = new HashSet<>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; 
      }    
   }		
}

Once constructed following platform api can be used to record these field operations:

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

LineageRecorder will be available in the initialize method of the CDAP programs through program runtime context.

Sample program code to record the operations as done for the above pipeline from ETLMapReduce. Note that this code is called from the CDAP program and not from the plugin.

public class ETLMapReduce extends AbstractMapReduce {
...
   public void initialize() throws Exception {
      MapReduceContext context = getContext();
	  ...
	  List<FieldOperation> operations = new ArrayList<>();
      outputSource source = new HashSet<>( Source.from("myns", "user_data", "files contains user information");
 
      inputFieldPathFieldOperation.Builder builder = new FieldPath(ArraysFieldOperation.asListBuilder("body"), source););
      builder
         input.addsetName(inputFieldPath);
 "read")
      outputFieldPath = new FieldPath(Arrays.asListsetDescription("first_name"), source);
  Reading the data from the file")
   output.add(outputFieldPath); 	  outputFieldPath = new FieldPath(Arrays.asList("last_name"), source);addInputSource(source)
         output.addaddOutput(outputFieldPath);
	offsetField)
      outputFieldPath = new FieldPath(Arrays.asList("age"), sourceaddOutput(bodyField);      
output.add(outputFieldPath); 	  outputFieldPath = new FieldPathoperations.add(Arraysbuilder.asListbuild("city"), source);
       
   output.add(outputFieldPath); 	  outputFieldPathbuilder = new FieldPath(ArraysFieldOperation.asListBuilder("state"), source););
      builder
         output.addsetName(outputFieldPath"parse");
	         operation = new FieldOperation.setDescription("parse", "parsing commathe separatedinput body field",)
 input, output);       operations.addaddInput(operationbodyField);
        input = new HashSet<>.addOutput(firstNameField);
      output = new HashSet<>.addOutput(lastNameField);
       inputFieldPath = new FieldPath(Arrays.asList("first_name"), source);addOutput(ageField)
         input.addaddOutput(inputFieldPathcityField);
      inputFieldPath = new FieldPath(Arrays.asList("last_name"), source)addOutput(stateField);
      inputoperations.add(inputFieldPathbuilder.build());
 

    	  outputFieldPathbuilder = new FieldPath(ArraysFieldOperation.asListBuilder("name"), source););
      builder
         output.addsetName(outputFieldPath);
 "concat")
      operation = new FieldOperation.setDescription("concat", "concatenatingconcatenating the first_ name and last_ name", input, output);
      operations.add(operation);        input = new HashSet<>();.addInput(firstNameField)
      output = new HashSet<>.addInput(lastNameField);
       inputFieldPath = new FieldPath(Arrays.asList("first_name"), source.addOutput(nameField);
      inputoperations.add(builder.build(inputFieldPath));


    operation  builder = new FieldOperation("drop", "deleting first_name", input, null /* drop does not generate any output field */);.Builder();
      builder
         .setName("drop")
         operations.add(operation);
setDescription("deleting the first name field")
      input = new HashSet<>.addInput(firstNameField);
      output = new HashSet<>operations.add(builder.build());
 
      inputFieldPathbuilder = new FieldPath(ArraysFieldOperation.asListBuilder("last_name"), source);
      input.add(inputFieldPath);builder
      operation = new FieldOperation("drop", "deleting last_name", input, null /* drop does not generate any output field */);
      operations.add(operation);.setName("drop")
         .setDescription("deleting the last name field")
       input = new HashSet<>.addInput(lastNameField);
      output = new HashSet<>operations.add(builder.build());

      inputFieldPathbuilder = new FieldPath(ArraysFieldOperation.asListBuilder("name"), source););
      builder
  input.add(inputFieldPath);
	    outputFieldPath = new FieldPath(Arrays.asListsetName("idcreate"),
  source);       output.add(outputFieldPath);setDescription("creating unique id from the name field")
      operation = new FieldOperation("create", "created unique id from name", input, output);.addInput(nameField)
         .addOutput(idField)
      operations.add(operationbuilder.build());

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

Following section still need to be updated. Please do not review.

In order for plugins to be able to record this information at runtime we can update the StageContext interface with the following method.

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

This method will then be available to the plugins in the prepareRun method where field mutations can be recorded. Note that we do not need to expose the dataset name parameter in this method. Since plugins operate on the individual records and they do not have any information about the dataset where the current record will land up. However data pipeline app knows about the datasets used in the pipeline and also the information about the DAG. It is possible to figure out from DAG which stage will lead to which dataset. So data pipeline can figure out this information and call the platform method with dataset name as an argument and field mutations accumulated from the various stages.

Pipeline config changes

For some of the plugins such as Javascript transform or Python transform, the prepareRun method is written by the plugin developer, however transform method is supplied by the pipeline developer. Since pipeline developer knows more about the transform method and can provide the field level mutations better, we will need an ability to provide the mutations via pipeline config as well.

Following are the proposed changes for the pipeline configs - 

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

Scenarios for Schema propagation

1. Configure time schema: When the pipeline is configured with the static schema, prepareRun method of the plugin will have access the schema. In this case the plugin will be able to emit the required lineage information.
2. Dynamic schema through macro: It is possible that the schema is provided as a macro during configure time and the value for which is supplied through runtime arguments. In this case in order for schema to be available in the prepareRun method we might need some changes.[TBD]

3. Unknown schema: It is possible to implement the custom plugins which work independent of the schema. For example consider the following pipeline - 

   File-------->CSV Parser--------->Javascript Transform-------->Custom Table Sink

   Assume that the Javascript transform just looks for the fields in the input StructuredRecord which contains "secure" in it, for example "secure_id", "secure_account" and drops them. Custom table sink takes all the fields from the input StructuredRecord and writes to a Table dataset with each individual field corresponds to the column name. In this case the schema of the Table will be variable based on the individual record. In this case even the pipeline developer wont be able to correctly specify the field mutations.

Approach

Approach #1

Approach #2

API changes

New Programmatic APIs

New Java APIs introduced (both user facing and internal)

Deprecated Programmatic APIs

New REST APIs

Deprecated REST API

CLI Impact or Changes

• Impact #1
• Impact #2
• Impact #3

UI Impact or Changes

• Impact #1
• Impact #2
• Impact #3

Security Impact 

What's the impact on Authorization and how does the design take care of this aspect

Impact on Infrastructure Outages 

System behavior (if applicable - document impact on downstream [ YARN, HBase etc ] component failures) and how does the design take care of these aspect

Test Scenarios

Releases

Release X.Y.Z

Release X.Y.Z

Related Work

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

Future work