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 84 Next »

Goals

Checklist

  • User stories documented (Albert/Vinisha) 
  • User stories reviewed (Nitin)
  • Design documented (Shankar/Kashif)
  • Design reviewed (Terence/Andreas)
  • Feature merged ()
  • Examples and guides ()
  • Integration tests () 
  • Documentation for feature ()
  • Blog post

Use Cases

  1. A pipeline developer wants to create a pipeline that has several configuration settings that are not known at pipeline creation time, but that are set at the start of the each pipeline run. 
    1. A pipeline developer wants to create a pipeline that reads from a database source and writes to a table sink. He wants to configure the name of the database table and name of the table sink on a per run basis and he gives those values as input before starting the run.  
  2. A pipeline developer wants to create a pipeline with a custom action at the start of the run. The custom action based on some logic provides the name of the database to use as source and the name of the table to write in sink. The next stage in pipeline uses this information to read from the appropriate database source and write to the table sink.  

User Stories

  1. As a pipeline developer, I want to be able to configure a plugin property to some value that will be substituted each run based on the runtime arguments.
  2. As a pipeline operator, I want to be able to set arguments for the entire pipeline that will be used for substitution.
  3. As a pipeline operator, I want to be able to set arguments for a specific stage in the pipeline that will be used for substitution.
  4. As a plugin developer, I want to be able to write a code that is executed at the start of the pipeline and sets arguments for the rest of the run.

Design

Macros Syntax


Function Syntax : 
${macroFunction(macro)}
 
Shorthand notation:
${macro}
 
Example Usage: 
${secure(accessKey)} - get access key from secure store 
${logicalStartTime(timeFormat)} - apply time function on the timeFormat provided and use the value. 
 
The Default (shortHand) usage will substitute arguments using the following precedence:
Custom Action Workflow-Token > Runtime Arguments > Stored Preferences
 
Examples:
ipConfig: ${hostname}:${port}
JDBC connection string : jdbc:${jdbc-plugin}://${hostname}:${sql-port}/${db-name}
 
Using the expanded syntax allows additional logic to be applied to the macro arguments through a macro function.
Escaping can be supported using the \ (backslash) character (e.g. \${hostname} will not be substituted)
 
Nested macros:
if a macro contains another macro, Example : ${secure(${user-name})} 
In the above example, we want to lookup the user-name in properties first, then use secure store to get the key/password for that user-name. this final key/password will be used for that field. 

The shorthand notation supports retrieval precedence to limit the exposure of underlying workflow-tokens and runtime-arguments to pipeline operators. The "functionTime" macro function uses the logical start time of a run to perform the substitution. This is an example of a macro function that is not just a key-value lookup but allows for extra logic to be performed before a value is returned. For now, the implementation will only support the following macro functions: runtime-arguments. Once the secure store API is available, it will also support secure store. In the future, we can see if we will allow developers to create custom macro functions (similar to functionTime(...)).


Notes:

  • The current implementation for macro substitution supports recursive expansion. That is, if a macro such as ${address} expands to ${hostname}:${port}, then ${hostname} and ${port} will be evaluated. However, this can lead to an infinite loop from circular macros, so we can add a maximum depth for expansion.

 


Pipeline Config
"stages": [
    {
        "name": "Database",
        "plugin": {
            "name": "Database",
            "type": "batchsource",
            "properties": {
				...
                "user": "${username}",
                "password": "${secure(sql-password)}",
                "jdbcPluginName": "jdbc",
                "jdbcPluginType": "${jdbc-type}",
                "connectionString": "jdbc:${jdbc-type}//${hostname}:${port}/${db-name}",
                "importQuery": "select * from ${table-name};"
            }
        }
    },
    {
        "name": "Table",
        "plugin": {
            "name": "Table",
            "type": "batchsink",                                        
            "properties": {
                "schema": "{\"type\":\"record\",\"name\":\"etlSchemaBody\",
                \"fields\":[{\"name\":\"name\",\"type\":\"string\"},
                {\"name\":\"age\",\"type\":\"int\"},{\"name\":\"emp_id\",\"type\":\"long\"}]}",
                "name": "${table-name}",
                "schema.row.field": "name"
            }
        }
    }
]

 

CDAP API Changes

Macro Annotation:

Plugins can use an "@Macro" annotation to specify if a plugin field can be a macro.  

When a plugin instance is instantiated at configure time, macros cannot be substituted as the values to substitute have not been specified yet. By default, macros will be disabled for properties. This is to prevent new plugin developers from having to worry about undefined behavior if they did not consider or are not familiar with macros.

 

Macro Annotation
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
/**
 * Annotation for a plugin field to specify if it can support macro
 */
public @interface Macro {
}


/**
 * Contains information about a property used by a plugin.
 */
@Beta
public class PluginPropertyField {

  private final String name;
  private final String description;
  private final String type;
  private final boolean required;
  // returns true if this field can accept macro
  private final boolean isMacro;
  ...
}


Example Macro Usage in Plugin:

public class TableSinkConfig extends PluginConfig {
  @Name(Properties.Table.NAME)
  @Description("Name of the table. If the table does not already exist, one will be created.")
  // The name of the table can be specified by a runtime macro, by default macros are disabled for fields.
  @Macro 
  private String name;

  @Name(Properties.Table.PROPERTY_SCHEMA)
  @Description("schema of the table as a JSON Object. If the table does not already exist, one will be " +
    "created with this schema, which will allow the table to be explored through Hive. If no schema is given, the " +
    "table created will not be explorable.")
  @Nullable
  private String schemaStr;

  @Name(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD)
  @Description("The name of the record field that should be used as the row key when writing to the table.")
  private String rowField;
}


PluginConfig Changes:

@Beta
public abstract class PluginConfig extends Config implements Serializable {

  /**
   * Returns the {@link PluginProperties}.
   */
  public final PluginProperties getProperties() {
    return properties;
  }
 
  /**
   * Returns false if the field is not annotated as a macro. If field is annotated as macro, 
   * the field value is checked if its a macro, based on the check it returns true/false.  
   */
   public final boolean isMacro(String fieldName) {
       ...
   }
}

The method will return whether or not the property with the provided fieldName contains a macro at configure time. We don't want to force deferring macros to runtime in the case that a field is macroable but actually has no macro provided in its configuration. This allows optional checking of properties at configure time for simple pipelines.  

 

Example usage
private final TableSinkConfig sinkConfig;
@Override
void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  if (!sinkConfig.isMacro("name")) {
 	// create dataset if the datasetName field is not a macro
    pipelineConfigurer.createDataset(name, datasetType, DatasetProperties.builder().addAll(properties).build());
  }
  ...
}


Notes:

  • To allow macro substitution of non-String properties, any properties configured with a macro will have a placeholder default value at configure time which is different from runtime.
  • For primitive types, this would be Java's default value, Example: integer field will have a value of 0. objects, would be null.
  • This is fine because we are exposing a method to check whether a field is a macro or not in pipelineConfigurer. so plugin developers can check at configure time if a field is macro before performing appropriate action using that field. 
  • During runtime, macro fields will be substituted and plugin instance will have fields with substituted values.


PluginContext Changes:

 

@Beta
public interface PluginContext {
 
/**
 * Creates a new instance of a plugin. The instance returned will have the {@link PluginConfig} setup with
 * {@link PluginProperties} provided at the time when the
 * {@link PluginConfigurer#usePlugin(String, String, String, PluginProperties)} was called during the
 * program configuration time. If a plugin field has a macro, 
 * the parameter evaluator is used to evaluate the macro and the evaluated value is used for the plugin field.
 *
 * @param pluginId the unique identifier provide when declaring plugin usage in the program.
 * @param evaluator the macro evaluator that's used to evaluate macro for plugin field if macro is supported on those fields. 
 * @param <T> the class type of the plugin
 * @return A new instance of the plugin being specified by the arguments
 *
 * @throws InstantiationException if failed create a new instance
 * @throws IllegalArgumentException if pluginId is not found
 * @throws UnsupportedOperationException if the program does not support plugin
 */
<T> T newPluginInstance(String pluginId, MacroEvaluator evaluator) throws InstantiationException;
MacroEvaluator
@Beta
interface MacroEvaluator {  
 
/**
 * lookup the property and return the value corresponding to the property.
 */ 
 String evaluate(String property);  
/**
 * Use the macro function and call the function with provided arguments, 
 * apply appropriate function or use properties in default case and return the evaluated value
 */ 
 String evaluate(String macroFunction, String... arguments);  
} 

Notes:
  • During configure time in the configurePipeline method, if a field is macro enabled, the property should not be validated as the macro has not been provided a substitutable value.
  • During runtime, PluginInstantiator has to understand if macro is enabled for a field, it parses the macro and then uses the parameter `evaluator` to evaluate the macro. 
  • The MacroEvaluator is provided by the DataPipelineApp. Though the platform understands macro, MacroEvaluator enables the actual substitution logic to be provided by the Application. 


ETL API - BatchContext Changes

Given that a dataset could be created at configure time if no fields are provided macros, a check should be available for plugin developers to see whether the dataset already exists at runtime. We can do this by altering the runtime context object passed in to the prepareRun method. As the object extends BatchContext which extends DatasetContext, we can create a new method in BatchContext that checks for the existence of a dataset.

This method will return whether or not the dataset with the provided datasetName already exists and can be used in prepareRun:

If The dataset to write to is macro-substituted and a macro is using in the config, we have to defer dataset creation to prepareRun rather than doing this in the configure stage. Deferring dataset creation in prepareRun will require adding a new method to BatchContext. 

@Beta
public interface BatchContext extends DatasetContext, TransformContext {

/**
 * create dataset identified by datasetName, typeName with properties.
 */
void createDataset(String datasetName, String typeName, DatasetProperties properties);

/**
 * return true if dataset with datasetName exists 
 */
boolean datasetExists(String datasetName);

...
}
@Override
public void prepareRun(BatchSinkContext context) {
  if (!context.datasetExists(config.getName())) {
    pipelineConfigurer.createDataset(config.getName(), ...);
  }
  // ...
}

Notes

Currently if a stream given in stream source or table given in table source doesn't exist, we create a new stream/table. We want to allow table creation as we want to create external dataset for sources, but disallow stream creation, so we are adding only createDataset to the BatchContext.

However there are certain fields which are used to determine the schema in the plugin and those cannot be macro-substituted as schema validation is essential during configure time and we want to disallow macro usage for them. 

Custom Action Setting Config Values:

One use case of the feature is to allow custom actions that run before a plugin to set macros. Custom actions can use workflow tokens to set values for field names. since workflow tokens are merged with runtime arguments and exposed in macro property lookup, macro substitution has access to tokens set from custom action, workflow tokens has higher priority over runtime arguments and preferences.  

Scoping (Low priority):

Scoping is currently at low priority and can be done manually. In our example config from a JDBC source to a table sink, there is a common macro "${table-name}", if the user wants to provide a different name for the table-name in Table Sink, he can manually do this:

SyntaxMacroEvaluates To
${table-name}table-nameemployees
${DBSource.table-name}DBSource.table-nameemployee_sql

This is more of the user creating unique argument keys as opposed to scoping.

Documentation/Changes

Regardless of where the substitution occurs, the guidelines for creating Hydrator plugins would have to change. For existing plugins, any validation for properties that are macro-substitutable in configurePipeline must be moved to prepareRun (see reference section for specific plugins). We also must document the convention for nulling/defaulting macroable properties at configure time.

Implementation Details

 

MacroContext
interface MacroContext {	
	/**
	 * Given the macro key, return the substituted value
     */ 
	String getValue(String macroKey);
}

 

 

Macro Types
Based on the macro type, one of the below MacroContext's will be used to get the value for macro. 
 
DefaultMacroContext implements MacroContext {
	Map<String, String> runtimeArguments;
	String getValue(String macroKey) {
		return runtimeArguments.get(macroKey);
	}
}

SecureMacroContext implements MacroContext {
	SecureStore secureStore;
	String getValue(String macroKey) {
		return secureStore.get(macroKey);
	}
}

RuntimeFunctionMacro implements MacroContext {	
	long logicalStartTime;
	Function<String, String> timezoneFunction;
	String getValue(String arguments) {
		return timezoneFunction.apply(arguments);
	}
} 


JIRA:

Error rendering macro 'jira' : Unable to locate Jira server for this macro. It may be due to Application Link configuration.

Reference

Many plugins have properties that are used in constructing or validating a schema at configure time. These fields need to have macros disabled to allow this. The following plugins and fields would be affected:

 

PluginFieldsUseConflict
BatchCassandraSourceschemaParsed for correctness to create the schema.Parsing a macro or schema with a nested macro would fail.
CopybookSourcecopybookContents

Copybook contents are converted to an InputStream and used to get external records, which are in turn used to add fields to the schema.

Schema would add macro literal as a field.
DedupAggregatoruniqueFields, filterOperationBoth fields are used to validate the input schema created.Macro literals do not exist as fields in schema and will throw IllegalArgumentException.
DistinctAggregatorfieldsSpecifies the fields used to construct the output schema.Will add macro literals as schema fields.*
GroupByAggregatorgroupByFields, aggregates,Gets fields from input schema and adds aggregates to to output fields list.Macro literals do not exist in input schema or are valid fields for an output schema.
RowDenormalizerAggregatorkeyField, nameField, valueFieldGets schemas by field names from the input schema.Macro literals do not exist as fields in the input schema.
KVTableSinkkeyField, valueFieldValidates that presence and type of these fields in the input schema.Macro literals will not exist in the input schema.
SnapshotFileBatchAvroSinkschemaParses schema to add file properties.Macro literals may disallow schema parsing or incorrect schema creation.
SnapshotFileBatchParquetSinkschemaParses schema to add file properties.Macro literals may disallow schema parsing or incorrect schema creation.
TableSinkschema, rowFieldValidates output and input schemas if properties specified.Macro literals will lead to failed validation of schema and row field.
TimePartitionedFileSetDatasetAvroSinkschemaParses schema to add file properties.Parsing macro literals in schema would fail.
TimePartitionedFileSetDatasetParquetSinkschemaParses schema to add file properties.Parsing macro literals in schema would fail.
SnapshotFileBatchAvroSourceschemaParses schema property to set output schema.Macro literals can lead to invalid schema parsing or creation.
SnapshotFileBatchParquetSourceschemaParses schema property to set output schema.Macro literals can lead to invalid schema parsing or creation.
StreamBatchSourceschema, name, formatStream is added and created through name and schema is parsed to set output schema.Macro literals will lead to bad parsing of properties.
TableSourceschemaSchema parsed to set output schema.Macro literals will lead to failed or incorrect schema creation.
TimePartitionedFileSetDatasetAvroSourceschemaSchema parsed to set output schema.Macro literals will lead to failed or incorrect schema creation.
TimePartitionedFileSetDatasetParquetSourceschemaSchema parsed to set output schema.Macro literals will lead to failed or incorrect schema creation.
JavaScriptTransformschema, script, lookupSchema format is used to set the output schema. JavaScript and lookup properties are also parsed for correctness.Macro literals can cause parsing to fail for schema creation, JavaScript compilation, or lookup parsing.
LogParserTransforminputNameGets field from input schema through inputName property.With a macro literal, the field will not exist in the input schema.
ProjectionTransformfieldsToKeep, fieldsToDrop, fieldsToConvert, fieldsToRenameProperties are used to create output schema.Macro literals will lied to a failed or wrong output schema being created.
PythonEvaluatorschemaSchema parsed for correctness and set as output schema.Macro literal can lead to failed or bad schema creation.
ValidatorTransformvalidators, validationScript,Validator property used to set validator plugins. Script property is also parsed for correctness.Macro literals can lead to failed parsing or plugins being set. Scripts can not be validated without validators.
ElasticsearchSourceschemaSchema parsed for correctness and set as output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
HBaseSinkrowField, schemaParsed to valid the output and input schemas and set the ouput schema.Macro literals can lead to failed or incorrect schema parsing/creation.
HBaseSourceschemaParsed for correctness to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
HiveBatchSourceschemaParsed for correctness to set ouput schema.Macro literals can lead to failed or incorrect schema parsing/creation.
MongoDBBatchSourceschemaParsed for correctness and validated to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
NaiveBayesClassifierpredictionFieldConfigures and sets fields of output schema and checked for existence in input schema.Output schema would be created wrongly with macro literal as prediction field and input schema check behavior is undefined.
Compressorcompressor, schemaParsed for correctness and used to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
CSVFormatterschemaParsed for correctness and used to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
CSVParserfieldValidated against input schema to check existence of field.Macro literals may not exist as fields in the input schema.
Decoderdecode, schemaDecode property is parsed and validated then used to validate the input schema. Schema parsed to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation or incorrect validation of input schema.
Decompressordecompressor, schemaDecompressor property is parsed and validated then used to validate the input schema. Schema parsed to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation or incorrect validation of input schema.
Encoderencode, schemaEncode property is parsed and validated then used to validate the input schema. Schema parsed to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation or incorrect validation of input schema.
JSONFormatterschemaParsed for correctness and used to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.
JSONParserfield, schemaValidates if field property is present in input schema. Parses schema property to set output schema.Macro literal may not exist in input schema and may lead to failed parsing or creation of output schema.
StreamFormatterschemaParsed for correctness and used to set output schema.Macro literals can lead to failed or incorrect schema parsing/creation.

 

May need verification

 


 

Other plugins have fields that are validated/processed at configure time that do not affect the schema. In these cases, these can be moved to the prepare run method. The following plugins and fields would be affected:

 

PluginFieldsUseJustification
StreamBatchSourceduration, delayParsed and validated for proper formatting.The parsing/validation is not related to the schema's creation.
TimePartitionedFileSetSourceduration, delayParsed and validated for proper formatting.The parsing/validation is not related to the schema's or dataset's creation.
ReferenceBatchSinkreferenceNameVerifies reference name meets dataset ID constraints.As dataset names can be macros, this supports the primary use case.
ReferenceBatchSourcereferenceNameVerifies that reference name meets dataset ID constraints.As dataset names can be macros, this supports the primary use case.
FileBatchSourcetimeTableCreates dataset from time table property.This is a primary use case for macros.
TimePartitionedFileSetSourcename, basePathName and basePath are used to create the dataset.This is a primary use case for macros.
BatchWritableSinkname, typeCreates dataset from properties.This is a primary use case for macros.
SnapshotFileBatchSinknameCreates dataset from name field.This is a primary use case for macros.
BatchReadableSourcename, typeDataset is created from name and type properties.This is a primary use case for macros.
SnapshotFileBatchSourceall properties*Creates dataset from properties.This is a primary use case for macros.
TimePartitionedFileSetSinkall properties*Creates dataset from properties.This is a primary use case for macros.
DBSourceimportQuery, boundingQuery, splitBy, numSplitsValidate connection settings and parsed for formatting.The parsing/validation does not lead to the creation of any schema or dataset.
HDFSSinktimeSuffixParsed to validate proper formatting of time suffix.The parsing/validation does not lead to the creation of any schema or dataset.
KafkaProducerasyncParsed to check proper formatting of boolean.The parsing/validation does not lead to the creation of any schema or dataset.
NaiveBayesClassifierfieldToClassifyChecked if input schema field is of type String.The validation does not lead to the creation or alteration of any schema.
NaiveBayesTrainerfieldToClassify, predictionFieldChecked if input schema fields are of type String and Double respectively.The validation does not lead to the creation or alteration of any schema.
CloneRecordcopiesValidated against being 0 or over the max number of copies.The validation does not lead to the creation of any schema or dataset.
CSVFormatterformatValidated for proper formatting.The validation does not lead to the creation of any schema or dataset.
CSVParserformatValidated for proper formatting.The validation does not lead to the creation of any schema or dataset.
HasherhashChecked against valid hash formats.The check does not lead to the validation or alteration of any schema.
JSONParsermappingMappings extracted and placed into a map with their expressions.The extraction does not affect any schema creation or validation.
StreamFormatterformatChecked against valid stream formats.The check does not lead to the validation or alteration of any schema.
ValueMappermapping, defaultsParsed after configuration is initialized and validated.The check does not lead to the validation or alteration of any schema.

 

May need verification


 

 

  • No labels