如何在Apache NiFi中自定义开发PutInfluxdb处理器？

没问题，我来一步步带你实现一个能把数据推送到InfluxDB的自定义NiFi处理器——只要遵循NiFi的开发规范，结合InfluxDB的Java客户端，整个过程其实很清晰。下面是详细步骤：

一、前置准备 #

• 确保你有**JDK 11+**的开发环境（NiFi 1.15+版本推荐用JDK 11，低版本可对应调整）
• 熟悉NiFi处理器的基本开发逻辑：核心是继承AbstractProcessor类，实现属性定义、关系定义和触发逻辑
• 确定你的InfluxDB版本：v1.x和v2.x的Java客户端差异较大，v1用influxdb-java依赖，v2用influxdb-client-java依赖

二、创建处理器项目骨架 #

用NiFi官方的Maven Archetype快速生成处理器bundle项目，执行以下命令（替换你的NiFi版本为实际使用的版本，比如1.23.2）：

mvn archetype:generate -DarchetypeGroupId=org.apache.nifi -DarchetypeArtifactId=nifi-processor-bundle-archetype -DarchetypeVersion=你的NiFi版本

生成项目后，找到processor模块，在其中创建你的自定义处理器类（比如PutInfluxDB.java）。

三、编写处理器核心逻辑 #

1. 定义处理器配置属性 #

首先定义连接InfluxDB所需的配置项，比如URL、令牌/用户名密码、组织/数据库等，用PropertyDescriptor来封装：

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

import java.util.ArrayList;
import java.util.List;

public class PutInfluxDB extends AbstractProcessor {

    // 示例：InfluxDB v2.x的核心属性
    public static final PropertyDescriptor INFLUXDB_URL = new PropertyDescriptor.Builder()
            .name("InfluxDB URL")
            .description("InfluxDB实例的访问URL，比如http://localhost:8086")
            .required(true)
            .addValidator(StandardValidators.URL_VALIDATOR)
            .build();

    public static final PropertyDescriptor INFLUXDB_TOKEN = new PropertyDescriptor.Builder()
            .name("InfluxDB Token")
            .description("InfluxDB v2.x的认证令牌")
            .required(true)
            .sensitive(true)
            .build();

    public static final PropertyDescriptor INFLUXDB_ORG = new PropertyDescriptor.Builder()
            .name("Organization")
            .description("InfluxDB v2.x的组织名称")
            .required(true)
            .build();

    public static final PropertyDescriptor INFLUXDB_BUCKET = new PropertyDescriptor.Builder()
            .name("Bucket")
            .description("InfluxDB v2.x的桶名称")
            .required(true)
            .build();

    // 重写方法返回所有支持的属性
    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        List<PropertyDescriptor> properties = new ArrayList<>();
        properties.add(INFLUXDB_URL);
        properties.add(INFLUXDB_TOKEN);
        properties.add(INFLUXDB_ORG);
        properties.add(INFLUXDB_BUCKET);
        return properties;
    }

2. 定义处理器关系 #

定义流文件处理后的流转方向：成功和失败：

// 关系定义
    public static final Relationship REL_SUCCESS = new Relationship.Builder()
            .name("success")
            .description("成功写入InfluxDB的流文件")
            .build();

    public static final Relationship REL_FAILURE = new Relationship.Builder()
            .name("failure")
            .description("写入InfluxDB失败的流文件")
            .build();

    @Override
    public Set<Relationship> getRelationships() {
        Set<Relationship> relationships = new HashSet<>();
        relationships.add(REL_SUCCESS);
        relationships.add(REL_FAILURE);
        return relationships;
    }

3. 实现核心触发逻辑 #

在onTrigger方法中完成流文件读取、InfluxDB客户端初始化、数据写入和结果流转：

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

        // 获取配置属性
        String url = context.getProperty(INFLUXDB_URL).getValue();
        String token = context.getProperty(INFLUXDB_TOKEN).getValue();
        String org = context.getProperty(INFLUXDB_ORG).getValue();
        String bucket = context.getProperty(INFLUXDB_BUCKET).getValue();

        InfluxDBClient client = null;
        WriteApi writeApi = null;

        try {
            // 初始化InfluxDB v2.x客户端
            client = InfluxDBClientFactory.create(url, token.toCharArray());
            writeApi = client.getWriteApi();

            // 读取流文件内容（假设是LineProtocol格式）
            try (InputStream in = session.read(flowFile)) {
                String lineProtocol = IOUtils.toString(in, StandardCharsets.UTF_8);
                // 写入数据
                writeApi.writeRecord(org, bucket, lineProtocol);
                getLogger().info("流文件 {} 成功写入InfluxDB", new Object[]{flowFile.getId()});
                // 流转到成功关系
                session.transfer(flowFile, REL_SUCCESS);
            }

        } catch (Exception e) {
            getLogger().error("流文件 {} 写入InfluxDB失败", new Object[]{flowFile.getId()}, e);
            // 流转到失败关系，并惩罚流文件
            session.transfer(flowFile, REL_FAILURE);
            session.penalize(flowFile);
        } finally {
            // 关闭资源
            if (writeApi != null) {
                writeApi.close();
            }
            if (client != null) {
                client.close();
            }
        }
    }
}

注意：如果是InfluxDB v1.x，替换客户端初始化逻辑为InfluxDB influxDB = InfluxDBFactory.connect(url, username, password);，然后用influxDB.write(database, retentionPolicy, Point)来写入数据。

四、打包部署 #

1. 在项目根目录执行Maven打包命令：

mvn clean package

2. 找到nifi-xxx-nar/target目录下生成的.nar文件（NiFi的插件包格式）
3. 将该nar包复制到NiFi安装目录的lib文件夹下
4. 重启NiFi服务，之后就能在NiFi UI的处理器列表中找到你的PutInfluxDB处理器了

五、测试验证 #

1. 拖入GenerateFlowFile处理器，配置生成LineProtocol格式的测试数据（比如cpu,host=server01 value=0.64 1434055562000000000）
2. 连接GenerateFlowFile到PutInfluxDB，配置好InfluxDB的连接参数
3. 启动处理器，查看InfluxDB是否收到数据，同时检查NiFi日志和流文件的流转状态

额外优化建议 #

• 批量写入：InfluxDB的WriteApi支持批量写入，可以收集多个流文件的内容批量提交，提升性能
• 参数验证：给属性添加更严格的验证器，比如检查令牌格式、组织/桶是否存在
• 重试机制：可以结合NiFi的重试策略，对失败的流文件进行自动重试
• 动态属性：支持动态设置measurement、tag等字段，让处理器更灵活

内容的提问来源于stack exchange，提问作者surya rahul