自定义Processor组件

public abstract class AbstractProcessor extends AbstractSessionFactoryProcessor {

    // 控制器是先调用的AbstractProcessor实现的这个onTrigger方法，然后再调用用户自定义实现的(下面的)onTrigger
    @Override
    public final void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
        final ProcessSession session = sessionFactory.createSession();
        try {
            onTrigger(context, session);
            session.commit();
        } catch (final Throwable t) {
            session.rollback(true);
            throw t;
        }
    }
    //这个onTrigger方法就是我们最常见的在自定义Processor里需要去实现功能逻辑的了
    public abstract void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException;
}

/**
 * 给简单的二级结构的json数据添加常量值
 *
 * @author 酷酷的诚
 * @date 2019-07-03 10:07
 */
@EventDriven
@SideEffectFree
@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"JSON", "Transform", "address"})
@CapabilityDescription("输入为json数组，为数组中的每一个元素增加常量")
public class JsonAddConstant extends AbstractJsonCleaningProcessor {
    public static final PropertyDescriptor CONSTANT_KEY = new PropertyDescriptor.Builder()
            .name("常量字段名")
            .description("增量常量的字段名称")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(true)
            .defaultValue("")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
    public static final PropertyDescriptor CONSTANT_VALUE = new PropertyDescriptor.Builder()
            .name("常量值")
            .description("增量的常量值")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(true)
            .defaultValue("")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final Set<Relationship> relationships = new HashSet<>();
        relationships.add(REL_SUCCESS);
        this.relationships = Collections.unmodifiableSet(relationships);
        final List<PropertyDescriptor> properties = new ArrayList<>();

        properties.add(CONSTANT_KEY);
        properties.add(CONSTANT_VALUE);

        this.properties = Collections.unmodifiableList(properties);
    }

    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }
        String constantKey = context.getProperty(CONSTANT_KEY).evaluateAttributeExpressions(flowFile).getValue();
        String constantValue = context.getProperty(CONSTANT_VALUE).evaluateAttributeExpressions(flowFile).getValue();

        ArrayNode arrayNode = validateAndEstablishJsonArray(session, flowFile);
        for (JsonNode jsonNode : arrayNode) {
            ((ObjectNode) jsonNode).put(constantKey, constantValue);
        }
        if (arrayNode.size() > 0) {
            // REL_SUCCESS
            session.transfer(session.write(flowFile, out -> out.write(arrayNode.toString().getBytes(StandardCharsets.UTF_8))), REL_SUCCESS);
        }
    }

/**
 * @author 酷酷的诚
 * @date 2019-06-20 13:59
 */
public abstract class AbstractJsonCleaningProcessor extends AbstractProcessor {
    public Set<Relationship> relationships;
    public List<PropertyDescriptor> properties;
    public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
            .description("清洗后数据路由到此关系").build();
    public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
            .description("清洗失败的数据路由到此关系").build();

    /**
     * flowfile转json node 并验证
     */
    public ArrayNode validateAndEstablishJsonArray(ProcessSession session, FlowFile flowFile) {
        final ObjectMapper mapper = new ObjectMapper();
        final AtomicReference<JsonNode> rootNodeRef = new AtomicReference<>(null);
        try {
            session.read(flowFile, in -> {
                try (final InputStream bufferedIn = new BufferedInputStream(in)) {
                    rootNodeRef.set(mapper.readTree(bufferedIn));
                }
            });
        } catch (final ProcessException pe) {
            getLogger().error("Failed to parse {} as JSON due to {}; routing to failure", new Object[]{flowFile, pe.toString()}, pe);
            session.transfer(flowFile, REL_FAILURE);
            return null;
        }
        final JsonNode rootNode = rootNodeRef.get();

        final ArrayNode dataJsonArray;
        if (rootNode.isArray()) {
            dataJsonArray = (ArrayNode) rootNode;
        } else {
            final JsonNodeFactory nodeFactory = JsonNodeFactory.instance;
            dataJsonArray = new ArrayNode(nodeFactory);
            dataJsonArray.add(rootNode);
        }
        return dataJsonArray;
    }

    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {

    }

    @Override
    public Set<Relationship> getRelationships() {
        return relationships;
    }

    @Override
    public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return properties;
    }
}