在Apache Nifi中创建自定义处理器

Question

我正在构建一个自定义处理器来处理流文件，为了处理流文件，我需要从我的本地文件系统读取CSV文件。我创建了一个proerty描述符CSV_PATH，如下所示

public static final PropertyDescriptor CSV_PATH = new 
PropertyDescriptor
.Builder().name("CSV Path")
.displayName("CSV Path")
.description("CSV Path Reader")
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();



@Override

protected void init(final ProcessorInitializationContext context) {

final List<PropertyDescriptor> descriptors = new 
ArrayList<PropertyDescriptor>();

descriptors.add(JSON_PATH);

descriptors.add(CSV_PATH);

this.descriptors = Collections.unmodifiableList(descriptors);



final Set<Relationship> relationships = new HashSet<Relationship>();

relationships.add(SUCCESS);

this.relationships = Collections.unmodifiableSet(relationships);

}

现在我想在配置处理器时获取在UI中设置的CSV_PATH属性的值。我无法获得CSV_PATH值。另外，如果我在代码中硬编码filepath，那么我仍然无法从本地文件系统读取CSV。

Answer 1

您希望使用以下代码从ProcessContext检索PropertyDescriptor的值：

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) {
    FlowFile flowFile = session.get();
    if (flowFile == null) {
        return;
    }

    final String csvPath = context.getProperty(CSV_PATH).getValue();

    // Do something with csvPath

}

如果您决定在属性描述符中支持NiFi表达式语言，则还需要对其进行评估：

最终字符串csvPath=context.getproperty(CSV_PATH).evaluateAttributeExpressions().getValue()；

这在Apache NiFi开发人员指南中有说明。我最近在巴塞罗那2019年Dataworks峰会上做了一次演示，介绍了定制处理器开发，包括一些最佳实践和可能有用的示例。您还可以查看NiFi代码库中的任何现有处理器以查看示例。