如何在Apache Camel中为现有CSV数据新增列和行并解决反序列化问题

问题原因说明 #

你当前拿到CsvIterator类型是因为配置了lazyLoad="true"，该配置会让camel-csv组件采用懒加载模式逐行读取CSV，避免大文件占满内存，该类实现了标准Iterator接口，并非无法使用，无需强转为其他类型。如果处理的CSV文件体积较小，直接将lazyLoad设置为false，反序列化后会直接返回List<Map<String, Object>>格式的全量CSV数据，操作更简便。

Bean组件实现方案 #

完全可以使用Bean组件完成CSV行列新增需求，完整实现流程如下：

1. 路由配置（XML格式） #

<!-- 读取CSV文件后反序列化 -->
<unmarshal>
    <csv delimiter="|" useMaps="true" lazyLoad="false" />
</unmarshal>
<!-- 调用自定义Bean处理CSV内容 -->
<bean ref="csvProcessBean" method="modifyCsv" />
<!-- 处理完成后序列化为CSV格式输出 -->
<marshal>
    <csv delimiter="|" headerDisabled="false" />
</marshal>

2. 自定义处理Bean代码 #

import org.apache.camel.Exchange;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class CsvProcessBean {
    public void modifyCsv(Exchange exchange) {
        // 获取反序列化后的CSV全量数据，每一个Map对应一行数据
        List<Map<String, Object>> csvRows = exchange.getIn().getBody(List.class);

        // 新增列：为所有行新增new_col列，可自定义赋值逻辑
        for (Map<String, Object> row : csvRows) {
            // 支持固定值赋值、基于已有列计算赋值等逻辑
            row.put("new_col", "自定义列值");
            // 示例：基于已有列计算 row.put("amount_tax", Integer.parseInt(row.get("amount").toString()) * 1.13);
        }

        // 新增行：构造新行数据加入列表
        Map<String, Object> newRow = new HashMap<>();
        newRow.put("原有表头1", "新行值1");
        newRow.put("原有表头2", "新行值2");
        newRow.put("new_col", "新行的自定义列值");
        csvRows.add(newRow);

        // 将修改后的数据写回消息体
        exchange.getIn().setBody(csvRows);
    }
}

大文件懒加载处理方案 #

如果需要处理大体积CSV必须保留lazyLoad="true"配置，按迭代器模式逐行处理即可，示例代码如下：

import org.apache.camel.Exchange;
import java.util.Iterator;
import java.util.Map;
import java.util.ArrayList;
import java.util.List;

public class LazyCsvProcessBean {
    public void modifyLazyCsv(Exchange exchange) {
        Iterator<Map<String, Object>> csvIterator = exchange.getIn().getBody(Iterator.class);
        List<Map<String, Object>> processedRows = new ArrayList<>();

        // 逐行读取处理
        while (csvIterator.hasNext()) {
            Map<String, Object> row = csvIterator.next();
            row.put("new_col", "懒加载处理列值");
            processedRows.add(row);
        }

        // 新增行逻辑和普通模式一致
        Map<String, Object> newRow = new HashMap<>();
        // 填充新行数据...
        processedRows.add(newRow);

        exchange.getIn().setBody(processedRows);
    }
}

注意事项 #

• 若原始CSV没有表头，需要在反序列化的csv标签中添加headers="表头1,表头2,表头3"属性指定表头，否则useMaps="true"配置下Map的key会默认使用列索引数字。
• 序列化CSV时如果不需要输出表头，将marshal中csv标签的headerDisabled属性设置为true即可。

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