Получить ProcessGroup от процессора - PullRequest
Новая
       11

Получить ProcessGroup от процессора

2 голосов
/ 21 марта 2019

Существует ли API для получения ProcessGroup по идентификатору от пользовательского процессора или процессора ExecuteScript?

Я знаю, что это возможно при использовании REST-API, но по соображениям безопасности у меня нет возможности использовать учетные данные для вызова API из службы.

Привет

1 Ответ

1 голос
/ 22 марта 2019

Если вы используете InvokeScriptedProcessor с использованием groovy, вы можете использовать 

context.procNode.processGroup

из ProcessContext. Если вы хотите извлечь всех родителей в виде хлебной крошки, вы можете использовать это: 

import groovy.json.*;
import org.apache.nifi.groups.*;

class GroovyProcessor implements Processor {

    def REL_SUCCESS = new Relationship.Builder()
        .name("success")
        .description('FlowFiles that were successfully processed are routed here').build()

    def ComponentLog log

    @Override
    void initialize(ProcessorInitializationContext context) { 
        log = context.logger 
    }

    @Override
    Set<Relationship> getRelationships() { 
        return [REL_SUCCESS] as Set 
    }

    void executeScript(ProcessSession session, ProcessContext context) {
        def flowFile = session.get()
        if (!flowFile) { return }

        def breadcrumb = getBreadCrumb(context.procNode.processGroup) + '->' + context.getName()
        flowFile = session.putAttribute(flowFile, 'breadcrumb', breadcrumb)

        // transfer
        session.transfer(flowFile, this.REL_SUCCESS) 
    }

    // Recursive funtion that gets the breadcrumb
    String getBreadCrumb(processGroup) {        
        def breadCrumb = ''
        if(processGroup.parent != null)
            breadCrumb = getBreadCrumb(processGroup.parent) + '->'

        return breadCrumb + processGroup.name
    }

    @Override
    void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException {  
        def session = sessionFactory.createSession()      
        try {
            executeScript( session, context)
            session.commit()
        } 
        catch (final Throwable t) {
            log.error('{} failed to process due to {}; rolling back session', [this, t] as Object[])
            session.rollback(true)
            throw t
        }
    }

    @Override
    PropertyDescriptor getPropertyDescriptor(String name) { null }

    @Override
    List<PropertyDescriptor> getPropertyDescriptors() { 
        return [] as List
     }


    @Override
    void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) { } 

    @Override
    Collection<ValidationResult> validate(ValidationContext context) { null }   

    @Override
    String getIdentifier() { null }    
}    

processor = new GroovyProcessor()
...