HCL Commerce Version 9.1.8.0 or later

Creando un procesador NiFi personalizado

Utilice los pasos siguientes para crear su nuevo procesador NiFi personalizado.

Procedure

  1. Cree las clases de procesador personalizadas y los casos de prueba.
    • Las clases para los procesadores se crean dentro del directorio /src/main/java/;
    • Los correspondientes casos de prueba del procesador se crean dentro del directorio /src/test/java/.
    1. Cree una clase en src/main/java y extiéndala con org.apache.nifi.processor.AbstractProcessor. Como alternativa, si desea ampliar las clases predeterminadas, amplíela utilizando com.hcl.software.data.iprocessor.processors.AbstractCommerceBaseProcessor .
    2. Proporcione los códigos y atributos adecuados utilizando las anotaciones siguientes. 
      @Tags({"example"})
@CapabilityDescription("Provide a descrption")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
    3. Para organizar las importaciones, haga clic con el botón derecho en el archivo Java y seleccione Origen > Organizar importaciones.
    4. Implemente el método heredado que falta. 
      @Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException 
{
// TODO Auto-generated method stub
}
      Ahora está preparado para empezar a escribir la lógica para el procesador personalizado.
  2. Cree la lógica personalizada para su procesador NiFi. Los siguientes pasos ilustran cómo la lógica que se crea en Java afecta al procesador NiFi, o se muestra en la interfaz de usuario NiFi.

    Consulte la Guía del Desarrollador NiFi para obtener más información sobre cómo crear un procesador NiFi personalizado, y para obtener una explicación exhaustiva de las API que se utilizan para desarrollar extensiones.

    1. Declare los campos de la propiedad utilizando org.apache.nifi.components. El tipo de datos del descriptor de propiedades aparece en la pestaña PROPIEDADES del procesador personalizado.

      Por ejemplo:

      public static final PropertyDescriptor SCROLL_DURATION = new PropertyDescriptor.Builder().name("Scroll Duration")
	   .description("The scroll duration is how long each search context is kept in memory")
	   .required(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	   .expressionLanguageSupported(true).build();
    2. Declare campos de relación utilizando org.apache.nifi.processor . El tipo de datos de relación aparece en la pestaña CONFIGURACIÓN del procesador personalizado.

      Por ejemplo:

      public static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder().name("success")
		     .description("The flow file with the specified content was successfully transferred").build();

	public static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder().name("failure")
		    .description("The flow file with the specified content has encountered an error during the transfer").build();
	
	public static final Relationship RELATIONSHIP_NEXT = new Relationship.Builder().name("next")
		    .description("The flow file with the specified content for the next iteration").build();
    3. Añada las propiedades y relaciones declaradas dentro del método init(). 
                   @Override 
             protected void init(final ProcessorInitializationContext context) {
		    getSupportedPropertyDescriptors().add(SCROLL_DURATION);
		    getRelationships().add(RELATIONSHIP_SUCCESS);
		    getRelationships().add(RELATIONSHIP_FAILURE);
		    getRelationships().add(RELATIONSHIP_NEXT);
    4. Anule el método onTrigger() para realizar las siguientes operaciones. 
            @Override
	public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
		
		final String duration = context.getProperty(SCROLL_DURATION).evaluateAttributeExpressions().getValue();
		
		try {
      1. Obtenga el FlowFile y su contenido de ProcessSession.
        final FlowFile flowFile = session.get();
			
final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
session.exportTo(flowFile, bytes);
bytes.close();
final String content =  bytes.toString();
      2. Realice las operaciones que el procesador debería hacer con el contenido.
        final Map<String, Object>  json = (Map<String, Object>) new JsonSlurper().parseText(content);
final String scroll = (String) json.get("_scroll_id");
final Map<String, Object> hits = (Map<String, Object>) json.get(FIELD_HITS);
final List<Object> docs = (List<Object>) hits.get(FIELD_HITS);
if (docs.size() > 0) {
    final String scrollBody = String.format("{ \"scroll\": \"%s\", \"scroll_id\": \"%s\" }", duration, scroll);
    FlowFile nextFlowFile = session.create(flowFile);
    session.putAttribute(nextFlowFile, "index.scroll.uri", "_search/scroll");
      3. Vuelva a escribir el resultado en flowFile.
        nextFlowFile = session.write(flowFile, new OutputStreamCallback() {
  @Override
  public void process(final OutputStream outputStream) throws IOException {
	 outputStream.write(scrollBody.getBytes());
	 outputStream.flush();
  }
});
session.getProvenanceReporter().modifyContent(nextFlowFile);
      4. Transfiera flowFile a la relación deseada.
        session.getProvenanceReporter().route(flowFile, RELATIONSHIP_SUCCESS);
session.transfer(flowFile, RELATIONSHIP_SUCCESS);
  3. En org.apache.nifi.processor.Processor, cree una entrada que contenga el nombre de clase completo del procesador personalizado recién creado. Si ha creado un servicio controlador personalizado, cree una entrada en org.apache.nifi.controlador.ControllerService.
    Por ejemplo, en el Explorador de proyectos, expanda commerce-custom-search-bundles > commerce-custom-search-proocessors > src/main/resources > META-INF > services y haga clic en org.apache.nifi.processor.Processor. Escriba el nombre de clase en el panel de entrada de texto en el lado derecho de la ventana.

Results

Se crea el procesador personalizado.