Customizing Ingest profiles

Ingest profiles are stored in ZooKeeper under the /ingestprofiles znode. A typical Ingest profile has a structure similar to the following example.

{
        "profileName": "DatabasePriceStage1a",
        "provider": [
            "com.mycompany.data.ingest.product.providers.ChangeSQL"
        ],
        "preprocessor": [],
        "postprocessor": []
    } 

Cutsomize the definition using the IndexSchemaProvider Java interface. The method is defined as follows.

public interface IndexSchemaProvider {
    public String invoke(final ProcessContext context, final ProcessSession session, final String schema)
    throws ProcessException;
}

This method provides five parameters:

context

Provides access to convenience methods for obtaining property values, delaying the scheduling of the pncessor, provides access to Controller Services, etc.

session

Provides access to a {@link ProcessSession}, which can be used for accessing FlowFiles, etc.

schema

Contains the Elasticsearch index schema setting and mapping definition.

return

The modified schema definition.

throws

Throws a ProcessExeption if processing did not complete normally.

As an example, suppose you want to customize a product index schema provider. Create the following class:

public class MyProductSchemaProvider implements IndexSchemaProvider {

    @override
    public String invoke(final ProcessContext context, final ProcessSession session, final String schema) throws ProcessException {

    String mySchema = schema;

    // TODO: logic for customizing schema definition goes here

    return mySchema;
  }
}

You can then invoke the class inside the ingest profile for the Index schema.

{
    "profileName": "MyProductSchema",
    "provider": [
        "com.mycompany.data.index.product.providers.MyProductSchemaProvider"
    ]
}

You can use the IngestExpressionProvider as a custom exit for customizing ETL. The interface is defined as follows.

public interface IngestExpressionProvider {
     public String invoke(final ProcessContext context, final ProcessSession session, final String expression)
        throws ProcessException;

This method provides five parameters:

context

Provides access to convenient methods for obtaining property values, delaying the scheduling of the processor, and providing access to controller services, among others.

session

Provides access to a {@link ProcessSession}, which can be used to access FlowFiles and other processes.

expression

Contains the the database SQL expression to be executed.

return

The modified SQL expression.

throws

ProcessException if processing did not complete normally.

To intervene in the default Product Stage 1a SQL definition, you can, for example, create the following extension.

public class MyProductStage1aProvider implements IngestExpressionProvider {
    @override
    public String invoke(final ProcessContext context, final ProcessSession session, final String expression) throws ProcessException {
        String myExpression = expression;
        //Logic for customizing database SQL expression goes here
        return myExpression;
    }
}

Pre-processing occurs between the Extract and Transform stages. You can add the public interface IngestFlowPreProcessor to your profile, as a custom user exit before the flow transform happens at the current stage. The interface is defined as follows.

public interface IngestFlowPreProcessor {
    public Map<String, Object> invoke (final ProcessContent context, final ProcessSession session,
        final Map<String, Object> data) throws ProcessException;
}

This method provides five parameters:

context

Provides access to convenient methods for obtaining property values and delaying the scheduling of the processor, and provides access to controller and other services.

session

Provides access to a {@link ProcessSession}, which can be used to access FlowFiles and other services.

data

Contains an entry from the database result set.

return

The modified data for downstream processing.

throws

ProcessException if processing did not complete properly.

The following shows example pre-processor code that uses IngestFlowPreProcessor.

public class MyProductStage1aPreProcessor implements IngestFlowPreProcessor {

	@Override
	public Map<String, Object> invoke(final ProcessContext context, final ProcessSession session,
		final Map<String, Object> data) throws ProcessException {

		// TODO: logic for customizing database result entry goes here

              return myData;
	}
}

You can call the IngestFlowPostProcessor method as a custom user exit after the Transformation stage and before the Loading stage. The method is defined as follows.

public interface IngestFlowPostProcessor {
    public Map<String, Object> invoke(final ProcessContext context, final ProcessSession session,
        final Map<String, Object> data, final Map<String, Object> document) throws ProcessoException;

This method provides six parameters:

context

Provides access to convenient methods for obtaining property values and delaying the scheduling of the processor, and proides access to controller and other services.

session

Provides access to a {@link ProcessSession}, which can be used to access FlowFiles and other services.

data

Contains an entry from the database result set.

document

Contains a Json object representing the final document to be sent to Elasticsearch.

return

The modified data for downstream processing.

throws

ProcessException if processing did not complete properly.

public class MyProductStage1aPostProcessor implements IngestFlowPostProcessor {

    @override
    public Map<String, Object> invoke(final ProcessContext context, final ProcessSession session, 
			final Map<String, Object> data, final Map<String, Object> document) throws ProcessException {

		Map<String, Object> myDocument = new HashMap(document);

		// TODO: logic for customizing the Json document before sending to Elasticsearch
		//
		// For example ....

		final Map<String, String> x_field1 = new HashMap<String, String>();
		// Retrieve value of FIELD1 from database result entry
		final String FIELD1 = (String) data.get("FIELD1");
		// Insert value of FIELD1 into the resulting document to be sent to Elasticsearch
		if (StringUtils.isNotBlank(FIELD1)) {
			// Re-use default dynamic keyword field name "*.raw" (can be found from Product index schema)
			String field1 = FIELD1.trim();
			//
			// TODO: perform your own transformation of field1
			//
			x_field1.put("raw", field1);
			// x_field1.raw
			myDocument.put("x_field1", x_field1);
		}
		return myDocument;
	}
}

The general steps involved in building your customizations are:

• Set up a NiFi Toolkit development environment. For more information, see Setting up the HCL Commerce Developer Search environment.
• Develop and unit test your Ingest profile custom logics.
• Create your Ingest profile configuration and bind it against a connector pipe in a NiFi runtime environment.
• Deploy your Ingest profile custom extensions to your NiFi runtime environment for further testing.