Flink读取CSV时出现NUMERIC_VALUE_ILLEGAL_CHARACTER异常求助

你遇到的这个问题其实很常见——当CSV里的数值型字段（比如你的POJO里的两个Integer字段）为空时，Flink默认的CSV解析器不知道怎么把空字符串转换成Integer类型，就会抛出NUMERIC_VALUE_ILLEGAL_CHARACTER异常。而当所有字段都非空时，解析器能正常转换，所以程序能跑通。

下面给你几个实用的解决方案：

1. 手动映射处理空值（兼容性最好） #

放弃直接用pojoType自动映射，改用types指定所有字段类型，然后在map算子里手动处理每个字段的空值转换。这样你能完全控制解析逻辑：

env.readCsvFile(input)
   .ignoreFirstLine()
   // 按CSV字段顺序指定所有类型，把Integer类型先按String读进来
   .types(String.class, String.class, String.class, String.class, String.class, 
          String.class, String.class, String.class, String.class, String.class, 
          String.class, String.class, String.class, String.class, String.class, 
          String.class, String.class, String.class, String.class, String.class, 
          String.class, String.class, String.class, String.class, String.class, 
          String.class, String.class, String.class, String.class, String.class)
   .map(fields -> {
       MyPojo pojo = new MyPojo();
       // 逐个赋值，处理字符串字段
       pojo.setField0((String) fields[0]);
       pojo.setField1((String) fields[1]);
       // ... 中间20多个字符串字段依次赋值 ...
       
       // 处理第一个Integer字段（原第13个字段，索引12）
       String intField1Str = (String) fields[12];
       pojo.setYourFirstIntField(intField1Str == null || intField1Str.isEmpty() ? null : Integer.parseInt(intField1Str));
       
       // 处理第二个Integer字段（原第26个字段，索引25）
       String intField2Str = (String) fields[25];
       pojo.setYourSecondIntField(intField2Str == null || intField2Str.isEmpty() ? null : Integer.parseInt(intField2Str));
       
       // 剩下的字符串字段继续赋值
       pojo.setField26((String) fields[26]);
       // ... 直到最后一个字段
       return pojo;
   });

2. 配置CSV解析器的空值规则（Flink 1.10+推荐） #

如果你的Flink版本是1.10及以上，可以直接配置GenericCsvInputFormat的空值标识，让解析器自动把空字符串转换成null，适配你的Integer类型字段：

import org.apache.flink.api.common.io.GenericCsvInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.api.common.io.SimpleStringEncoder;

// 创建自定义CSV输入格式
GenericCsvInputFormat<MyPojo> csvInputFormat = new GenericCsvInputFormat<>(
    new Path(input),
    new SimpleStringEncoder<>(),
    MyPojo.class,
    getFields(), // 你的字段名数组
    null, // 使用默认逗号分隔符
    null, // 无引号字符
    null, // 无注释前缀
    true, // 忽略首行
    false, // 非宽松模式（可选，根据需求调整）
    0, // 不跳过错误行
    new String[]{""}, // 将空字符串识别为null
    false // 不解析注释
);

// 用自定义格式创建数据源
DataSource<MyPojo> source = env.createInput(csvInputFormat);

关键注意事项 #

• 确保你的POJO里的数值字段是包装类（比如Integer），而不是基本类型（int）。因为基本类型不能为null，空值无法赋值，会导致转换失败。
• 如果CSV里的空字段是空白字符（比如空格）而不是空字符串，记得在转换前先调用trim()处理，比如intField1Str.trim().isEmpty()。

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