Java使用Jackson实现AVRO转CSV时报空指针与生成异常如何解决

解决方案 #

第一步：修复依赖与语法问题 #

1. 补充jackson-datatype-jsr310依赖版本，保持所有Jackson组件版本统一：

<dependency>
    <groupId>com.fasterxml.jackson.datatype</groupId>
    <artifactId>jackson-datatype-jsr310</artifactId>
    <version>${jackson.version}</version>
</dependency>

2. 修正EmployeeModel构造方法的语法错误，删掉多余逗号：

public EmployeeModel(String id, String name, ZonedDateTime modifiedTimestamp,
    @JsonProperty("score") String score) {
    this.id = id;
    this.name = name;
    this.modifiedTimestamp = modifiedTimestamp;
    this.score = score;
}

第二步：修正CSV序列化逻辑 #

不要手动创建CsvGenerator，直接通过ObjectWriter写入输出流，同时注册JSR310模块处理时间类型序列化，修改后的转换方法如下：

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.dataformat.csv.CsvMapper;
import com.fasterxml.jackson.dataformat.csv.CsvSchema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.file.DataFileReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.util.Collections;

public class Test {
    private File avroToCsv(File avroFile, String path) {
        File result = new File(path);
        if (result.exists()) {
            result.delete();
        }
        // 用try-with-resources自动关闭流，避免资源泄漏
        try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(avroFile, new GenericDatumReader<>());
             OutputStream outStream = new FileOutputStream(result, true)) {
            
            CsvMapper csvMapper = new CsvMapper();
            // 注册JSR310模块，支持时间类型序列化为字符串格式
            csvMapper.findAndRegisterModules();
            CsvSchema schema = csvMapper.schemaFor(EmployeeModel.class).withHeader();
            ObjectWriter csvWriter = csvMapper.writer(schema);
            
            // 写入表头
            csvWriter.writeValue(outStream, Collections.emptyList());
            
            GenericData.Record record;
            while (reader.hasNext()) {
                record = reader.next();
                // 从AVRO记录中取值，修正时间字段取值逻辑
                Object modifiedTimeVal = record.get("modifiedTimestamp");
                LocalDateTime dateTime = LocalDateTime.parse(modifiedTimeVal.toString());
                ZonedDateTime modifiedDate = ZonedDateTime.of(dateTime, ZoneId.systemDefault());
                EmployeeModel tempModel = new EmployeeModel(
                        record.get("id").toString(),
                        record.get("name").toString(),
                        modifiedDate,
                        record.get("score").toString()
                );
                // 直接写入输出流，不需要手动创建CsvGenerator
                csvWriter.writeValue(outStream, tempModel);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    public static void main(String []args) {
        new Test().avroToCsv(new File("abc.avro"), "c:/test/ModifiedUsers.csv");
    }
}

提示：如果AVRO文件中modifiedTimestamp字段存储的是毫秒时间戳，可将时间转换逻辑替换为：

long timestamp = (long) record.get("modifiedTimestamp");
LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault());

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