如何使用 Spring 批处理 FlatFileItemReader 仅读取 CSV 文件中的子集列？

Question

我有一个读者知道如何读取包含 14 列的 CSV 文件，我希望它能够接收包含多列 (~500) 的文件并且只读取这 14 列，我知道解决方案应该包括 FieldSetMapper （根据这个问题：read only selective columns from csv file using spring batch）但我找不到合适的例子。这是我目前的读者：

@Bean
public FlatFileItemReader<RowInput> csvRowsReader() {
    FlatFileItemReader<RowInput> reader = new FlatFileItemReader<>();
    Resource resource = new FileSystemResource(new File(FileManager.getInstance().getInputFileLocation()));
    reader.setResource(resource);

    reader.setLinesToSkip(1);
    reader.setLineMapper(new DefaultLineMapper<RowInput>(){{
        setLineTokenizer(new DelimitedLineTokenizer(){{
            setNames(new String[]{"Field_1", "Field_2", "Field_3", "Field_4", "Field_5",
                    "Field_6", "Field_7", "Field_8", "Field_9", "Field_10", "Field_11",
                    "Field_12", "Field_13", "Field_14"});
        }});
        setFieldSetMapper(new BeanWrapperFieldSetMapper<RowInput>(){{
            setTargetType(RowInput.class);
        }});
    }});


    reader.setLinesToSkip(1);
    return reader;
}

例外是：

原因：org.springframework.batch.item.file.transform.IncorrectTokenCountException：记录中发现的令牌数不正确：预期为 14，实际为 544

我尝试使用的 FieldSetMapper：

    public class InputFieldSetMapper implements FieldSetMapper<RowInput>{

    public RowInput mapFieldSet(FieldSet fs) {

        if (fs == null) {
            return null;
        }

        RowInput input = new RowInput();
        input.setField1(fs.readString("Field_1"));
        input.setField2(fs.readString("Field_2"));
        // and so on...

        return input;
    }
}

Answer 1

您需要在LineTokenizer 上设置includedFields 属性，以指定在解析输入文件时要包含哪些字段。在你的情况下，它应该是这样的：

@Bean
public FlatFileItemReader<RowInput> csvRowsReader() {
    FlatFileItemReader<RowInput> reader = new FlatFileItemReader<>();
    Resource resource = new FileSystemResource(new File(FileManager.getInstance().getInputFileLocation()));
    reader.setResource(resource);

    reader.setLinesToSkip(1);
    reader.setLineMapper(new DefaultLineMapper<RowInput>(){{
        setLineTokenizer(new DelimitedLineTokenizer(){{
            setNames(new String[]{"Field_1", "Field_2", "Field_3", "Field_4", "Field_5",
                    "Field_6", "Field_7", "Field_8", "Field_9", "Field_10", "Field_11",
                    "Field_12", "Field_13", "Field_14"});
            setIncludedFields(0, 1, 2, 3, 4, 5, 6, 7, 8, 9,10 ,11 ,12 ,13);
        }});
        setFieldSetMapper(new BeanWrapperFieldSetMapper<RowInput>(){{
            setTargetType(RowInput.class);
        }});
    }});

    return reader;
}

编辑：添加一个非连续字段的示例

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;

@Configuration
@EnableBatchProcessing
public class MyJob {

    @Autowired
    private JobBuilderFactory jobs;

    @Autowired
    private StepBuilderFactory steps;

    @Bean
    public FlatFileItemReader<Person> itemReader() {
        return new FlatFileItemReaderBuilder<Person>()
                .name("personItemReader")
                .resource(new ClassPathResource("persons.csv"))
                .delimited()
                .includedFields(new Integer[] {0, 2})
                .names(new String[] {"id", "lastName"})
                .targetType(Person.class)
                .build();
    }

    @Bean
    public ItemWriter<Person> itemWriter() {
        return items -> {
            for (Person item : items) {
                System.out.println("person = " + item);
            }
        };
    }

    @Bean
    public Step step() {
        return steps.get("step")
                .<Person, Person>chunk(1)
                .reader(itemReader())
                .writer(itemWriter())
                .build();
    }

    @Bean
    public Job job() {
        return jobs.get("job")
                .start(step())
                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

    public static class Person {
        String id;
        String firstName;
        String lastName;
        int age;

        public Person() {
        }

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public String getFirstName() {
            return firstName;
        }

        public void setFirstName(String firstName) {
            this.firstName = firstName;
        }

        public String getLastName() {
            return lastName;
        }

        public void setLastName(String lastName) {
            this.lastName = lastName;
        }

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "id='" + id + '\'' +
                    ", firstName='" + firstName + '\'' +
                    ", lastName='" + lastName + '\'' +
                    ", age=" + age +
                    '}';
        }
    }

}

输入文件persons.csv如下：

1,foo1,bar1,10
2,foo2,bar2,20

该示例显示了如何仅映射 id 和 lastName 字段。

希望这会有所帮助。