NiFi中使用POI API转换CSV至Excel失败求助

我希望通过Groovy脚本结合POI API，将NiFi中CSV格式的FlowFile内容转换为XLS文件，目前使用Ivy获取POI依赖，依赖拉取看似正常。

NiFi流程 #

包含FlowFile获取、Groovy脚本执行、成功/失败分流分支的标准处理流程。

初始Groovy脚本 #

@Grapes(@Grab(group='org.apache.poi', module='poi-ooxml', version='3.9'))
import org.apache.poi.ss.usermodel.*
import org.apache.poi.hssf.usermodel.*
import org.apache.poi.xssf.usermodel.*
import org.apache.poi.ss.util.*
import org.apache.poi.ss.usermodel.*
import org.apache.poi.hssf.extractor.*
import java.nio.charset.*
import java.io.*
import org.apache.commons.io.IOUtils

def flowFile = session.get()
def date = new Date()

if(!flowFile) return

flowFile = session.write(flowFile, {inputStream, outputStream ->
try {
    
Workbook wb = WorkbookFactory.create(inputStream,);
Sheet mySheet = wb.getSheetAt(0);


def record = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
outputStream.write(record.getBytes(StandardCharsets.UTF_8))

wb.close();

} catch(e) {
 log.error("Error during processing", e)
 session.transfer(flowFile, REL_FAILURE)
}
} as StreamCallback)


def filename = flowFile.getAttribute('filename')+'.xlsx'
flowFile = session.putAttribute(flowFile, 'filename', filename) 

session.transfer(flowFile, REL_SUCCESS)

运行错误 #

执行后抛出POI格式识别异常，提示无法将CSV流解析为Excel文件。

修改后的脚本（仍报相同错误） #

@Grapes(@Grab(group='org.apache.poi', module='poi-ooxml', version='3.9'))
import com.opencsv.CSVReader
@Grapes(@Grab(group='com.opencsv', module='opencsv', version='4.2'))
import org.apache.poi.ss.usermodel.*
import org.apache.poi.hssf.usermodel.*
import org.apache.poi.xssf.usermodel.*
import org.apache.poi.ss.util.*
import org.apache.poi.ss.usermodel.*
import org.apache.poi.hssf.extractor.*
import java.nio.charset.*
import java.io.*
import org.apache.commons.io.IOUtils

def flowFile = session.get()
def date = new Date()

if(!flowFile) return

flowFile = session.write(flowFile, {inputStream, outputStream ->
try {
def nextLine = ''
reader = new CSVReader(new FileReader(csvFilePath), FILE_DELIMITER);

  Workbook wb = WorkbookFactory.create(inputStream,);
 Sheet sheet1 = wb.createSheet("Feuille");

// workBook = new SXSSFWorkbook();
//sheet = (SXSSFSheet) workBook.createSheet('Sheet');

int rowNum = 0;
while((nextLine = reader.readNext()) != null) {
Row currentRow = sheet1.createRow(rowNum++);
for(int i=0; i < nextLine.length; i++) {
if(NumberUtils.isDigits(nextLine[i])) {
currentRow.createCell(i).setCellValue(Integer.parseInt(nextLine[i]));
} else if (NumberUtils.isNumber(nextLine[i])) {
currentRow.createCell(i).setCellValue(Double.parseDouble(nextLine[i]));
} else {
currentRow.createCell(i).setCellValue(nextLine[i]);
}
}
}

fileOutputStream = new FileOutputStream(generatedXlsFilePath.trim());
wb.write(fileOutputStream);

wb.close();
fileOutputStream.close();
reader.close();
outputStream.close();

} catch(e) {
session.transfer(flowFile, REL_FAILURE)
}
} as StreamCallback)

def filename = flowFile.getAttribute('filename')+'.xlsx'
flowFile = session.putAttribute(flowFile, 'filename', filename)

session.transfer(flowFile, REL_SUCCESS)

错误原因分析 #

1. 核心逻辑错误：WorkbookFactory.create(inputStream)是POI用来读取已有Excel文件流的API，你传入的是CSV格式流，POI无法识别，直接抛出格式不匹配异常。
2. 修改后脚本的额外问题：
   • 仍保留了WorkbookFactory.create(inputStream)调用，重复触发格式错误；
   • 使用FileReader(csvFilePath)读取本地文件，但NiFi中FlowFile内容只能通过inputStream获取，且csvFilePath是未定义变量；
   • FILE_DELIMITER、generatedXlsFilePath均为未定义变量；
   • 使用FileOutputStream写入本地文件，违反NiFi的FlowFile处理模型，应直接写入提供的outputStream；
   • 未导入NumberUtils类，会触发类找不到异常；
   • 手动关闭outputStream，导致NiFi无法完成FlowFile的输出处理。

正确实现脚本 #

@Grapes([
    @Grab(group='org.apache.poi', module='poi-ooxml', version='3.9'),
    @Grab(group='com.opencsv', module='opencsv', version='4.2'),
    @Grab(group='org.apache.commons', module='commons-lang3', version='3.12.0')
])
import org.apache.poi.ss.usermodel.*
import org.apache.poi.xssf.usermodel.XSSFWorkbook
import com.opencsv.CSVReader
import org.apache.commons.lang3.math.NumberUtils
import java.nio.charset.StandardCharsets
import java.io.*

def flowFile = session.get()
if(!flowFile) return

try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        // 新建XLSX工作簿，而非读取现有文件
        Workbook wb = new XSSFWorkbook()
        Sheet sheet = wb.createSheet("Data")

        // 从FlowFile的输入流读取CSV内容
        CSVReader reader = new CSVReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8), ',')
        String[] nextLine
        int rowNum = 0

        while((nextLine = reader.readNext()) != null) {
            Row row = sheet.createRow(rowNum++)
            for(int i=0; i < nextLine.length; i++) {
                String cellValue = nextLine[i]
                if(NumberUtils.isDigits(cellValue)) {
                    row.createCell(i).setCellValue(Integer.parseInt(cellValue))
                } else if(NumberUtils.isNumber(cellValue)) {
                    row.createCell(i).setCellValue(Double.parseDouble(cellValue))
                } else {
                    row.createCell(i).setCellValue(cellValue)
                }
            }
        }

        // 将生成的Excel写入FlowFile输出流
        wb.write(outputStream)

        // 关闭资源（不关闭outputStream，由NiFi处理）
        reader.close()
        wb.close()
    } as StreamCallback)

    // 修正文件名：替换原后缀为xlsx，避免重复拼接
    def originalName = flowFile.getAttribute('filename')
    def newName = originalName.lastIndexOf('.') > -1 ? 
        originalName.substring(0, originalName.lastIndexOf('.')) + '.xlsx' : 
        originalName + '.xlsx'
    flowFile = session.putAttribute(flowFile, 'filename', newName)

    session.transfer(flowFile, REL_SUCCESS)
} catch(Exception e) {
    log.error("CSV转XLSX处理失败", e)
    session.transfer(flowFile, REL_FAILURE)
}

关键修正点 #

• 移除WorkbookFactory.create(inputStream)，改为新建XSSFWorkbook来创建空Excel文件；
• 通过InputStreamReader读取FlowFile的CSV输入流，避免本地文件依赖；
• 直接将Excel内容写入NiFi提供的outputStream，符合FlowFile处理模型；
• 补全NumberUtils的依赖导入，修复类找不到问题；
• 修正文件名逻辑，避免生成类似data.csv.xlsx的错误文件名；
• 仅关闭自定义创建的流，保留NiFi的outputStream由框架处理。

内容的提问来源于stack exchange，提问作者O. Sam