如何确保多个Apache NiFi处理器共享同一数据库连接会话？

针对你需要让步骤2-4（创建临时表、插入数据、调用函数并删表）处于同一数据库会话的需求，以下是两种可行的实现方式：

方式一：使用单处理器脚本整合所有操作 #

这是最直接且无需额外开发的方案，利用ExecuteGroovyScript或ExecuteScript处理器，在单个处理器内完成所有需要共享会话的操作：

• 配置处理器关联目标数据库的DBCPConnectionPool服务；
• 编写脚本逻辑：
  1. 从连接池获取数据库连接；
  2. 执行创建临时表的SQL语句；
  3. 读取上游FlowFile中的数据，解析后执行批量插入操作；
  4. 调用目标数据库的处理函数；
  5. 执行删除临时表的SQL；
  6. 提交事务（若操作成功）或回滚（若出现异常），最后释放连接回连接池。
• 示例Groovy脚本核心逻辑片段：

def conn = dbConnectionPool.getConnection()
try {
    def stmt = conn.createStatement()
    // 创建临时表
    stmt.execute("CREATE TEMPORARY TABLE temp_data (id INT, content VARCHAR(255))")
    // 插入数据（假设FlowFile内容为CSV格式）
    def flowFileContent = session.read(flowFile).getText()
    flowFileContent.split('\n').each { line ->
        def parts = line.split(',')
        stmt.executeUpdate("INSERT INTO temp_data VALUES (${parts[0]}, '${parts[1]}')")
    }
    // 调用处理函数
    stmt.execute("CALL process_temp_data()")
    // 删除临时表
    stmt.execute("DROP TABLE temp_data")
    conn.commit()
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    conn.rollback()
    session.transfer(flowFile, REL_FAILURE)
    log.error("操作失败: ${e.message}", e)
} finally {
    conn.close()
}

方式二：自定义处理器封装会话绑定逻辑 #

如果需要更灵活的控制或复用性，可以开发自定义NiFi处理器：

• 在处理器内部维护同一个数据库连接会话，依次执行步骤2-4的所有SQL操作；
• 利用NiFi的数据库连接池服务获取连接，全程保持连接打开状态直到所有操作完成；
• 处理器内部实现事务管理，确保操作的原子性。

关键注意事项 #

• 临时表的作用域是会话级，必须确保所有相关操作在同一个数据库连接内完成，否则无法访问临时表；
• 脚本或自定义处理器中必须处理异常回滚，避免因操作失败残留临时表；
• 使用连接池时，务必在操作完成后释放连接，防止连接泄漏导致连接池耗尽。

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