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

【问题标题】：Creating Custom Processor In Apache Nifi在 Apache Nifi 中创建自定义处理器
【发布时间】：2019-03-28 02:01:29
【问题描述】：

我正在构建一个自定义处理器来处理流文件，为了处理我需要从本地文件系统读取 CSV 文件的流文件。我创建了一个属性描述符 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 值。此外，如果我在代码中硬编码文件路径，那么我仍然无法从本地文件系统读取 CSV。

【问题讨论】：

标签： apache-nifi

【解决方案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 Expression Language，您还需要对此进行评估：

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

为此还有其他方法覆盖，包括流文件属性、变量注册表、自定义装饰器等。

这在Apache NiFi Developer's Guide 中有记录。我最近做了一个presentation at Dataworks Summit Barcelona 2019，涵盖了定制处理器开发，其中包括一些最佳实践和可能有用的示例。您还可以查看 NiFi 代码库中的任何现有处理器以查看示例。

【讨论】：