java - FlatFileItemWriter 无法创建新文件或更新现有文件

Question

[SpringBatch 新手] 使用 Spring Boot，我尝试创建一个作业，从 MongoDB 读取名称，转换为小写，然后输出到 CSV 文件。我的读取器和处理器正在工作，但写入器却没有。

我的代码如下。

配置文件:

package bbye;

import java.util.HashMap;
import java.util.Map;

import javax.sql.DataSource;

import org.springframework.batch.core.Job;
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.support.RunIdIncrementer;
import org.springframework.batch.item.data.MongoItemReader;
import org.springframework.batch.item.data.builder.MongoItemReaderBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.file.transform.FieldExtractor;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;

import hello.Person;

@Configuration
@EnableBatchProcessing
public class BatchConfigProcessing {


    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Autowired
    private MongoTemplate mongoTemplate;

    private String readQuery = "{}";

    // tag::readerwriterprocessor[]
    @Bean
    public MongoItemReader<Person> readMongo(DataSource dataSource) {
        return new MongoItemReaderBuilder<Person>()
                .name("mongoDocReader")
                .jsonQuery(readQuery)
                .targetType(Person.class)
                .sorts(sort())
                .template(mongoTemplate)
                .collection("people")
                .build();
    }

    @Bean
    public PersonDocProcessor processor() {
        return new PersonDocProcessor();
    }


    @Bean
    public FlatFileItemWriter<Person> writer() {
        /*FlatFileItemWriterBuilder<Person> writePerson =  new FlatFileItemWriterBuilder<Person>();
            writePerson.name("personDocWriter");
            writePerson.resource(new ClassPathResource("PersonExtracted.csv"));
            writePerson.lineAggregator(new DelimitedLineAggregator<Person>());
            writePerson.shouldDeleteIfExists(true);
            writePerson.build();*/

     FlatFileItemWriter<Person> fileWriter = new FlatFileItemWriter<>();
     fileWriter.setName("csvWriter");
     fileWriter.setResource(new ClassPathResource("PersonExtracted.csv"));
     fileWriter.setLineAggregator(lineAggregator());
     fileWriter.setForceSync(true);
     fileWriter.close();   
     return fileWriter; 
    }
    // end::readerwriterprocessor[]

    // tag::jobstep[]

    @Bean
    public Job exportUserJob(FileUploadNotificationListener listener, Step step1) {
        return jobBuilderFactory.get("exportUserJob")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build();
    }

    @Bean
    public Step step2(MongoItemReader<Person> reader) {
        return stepBuilderFactory.get("step2")
            .<Person, Person> chunk(10)
            .reader(reader)
            .processor(processor())
            .writer(writer())
            .build();
    }

    // end::jobstep[]

    public FieldExtractor<Person> fieldExtractor()
    {
        BeanWrapperFieldExtractor<Person> extractor = new BeanWrapperFieldExtractor<>();
        extractor.setNames( new String[] { "firstName",
                                            "lastName"});
        return extractor;

    }


    public LineAggregator<Person> lineAggregator() {
        DelimitedLineAggregator<Person> la = new DelimitedLineAggregator<Person>();
        la.setDelimiter(",");
        la.setFieldExtractor(fieldExtractor());
        return la;
    }

    public Map<String, Sort.Direction> sort(){
        String firstName = "firstName";
        Map<String, Sort.Direction> sortMap = new HashMap();
        sortMap.put(firstName, Sort.DEFAULT_DIRECTION);
        return sortMap;
    }


}

处理器文件

package bbye;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;

import hello.Person;

@Component
public class PersonDocProcessor implements ItemProcessor<Person, Person> {

    private static final Logger log = LoggerFactory.getLogger(PersonDocProcessor.class);

    @Override
    public Person process(final Person person) throws Exception {
        final String firstName = person.getFirstName().toLowerCase();
        final String lastName = person.getLastName().toLowerCase();

        final Person transformedPerson = new Person(firstName, lastName);

        log.info("Converting (" + person + ") into (" + transformedPerson + ")");

        return transformedPerson;
    }
}

监听器

package bbye;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.stereotype.Component;

@Component
public class FileUploadNotificationListener implements JobExecutionListener {

    @Override
    public void beforeJob(JobExecution jobExecution) {
        System.out.println("===== listening for job - mongoReader - fileWriter ====");

    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        System.out.println("==== file write job completed =====");

    }
}

这里的 Person 是一个简单的 POJO。手动创建文件和不创建文件的堆栈跟踪如下:

1. 如果 src/main/resources 下不存在该文件，FlatFileItemWriter 不会创建新文件并抛出异常

org.springframework.batch.item.ItemStreamException: Could not convert resource to file: [class path resource [PersonExtracted.csv]]
    at org.springframework.batch.item.file.FlatFileItemWriter.getOutputState(FlatFileItemWriter.java:399) ~[spring-batch-infrastructure-4.0.1.RELEASE.jar:4.0.1.RELEASE]
    at org.springframework.batch.item.file.FlatFileItemWriter.open(FlatFileItemWriter.java:337) ~[spring-batch-infrastructure-4.0.1.RELEASE.jar:4.0.1.RELEASE]
........
Caused by: java.io.FileNotFoundException: class path resource [PersonExtracted.csv] cannot be resolved to URL because it does not exist

如果我手动创建 PersonExtracted.csv 文件，程序运行时不会出现错误，但不会向该文件写入任何内容。事实上，返回的是一个空白文件。堆栈跟踪如下。

:: Spring Boot ::        (v2.0.2.RELEASE)

2018-06-19 11:35:17.663  INFO 25136 --- [           main] hello.Application                        : Starting Application on MyPC with PID 25136 (C:\eclipse-workspace\gs-batch-processing-master\complete\target\classes started by shristi in C:\eclipse-workspace\gs-batch-processing-master\complete)
2018-06-19 11:35:17.666  INFO 25136 --- [           main] hello.Application                        : No active profile set, falling back to default profiles: default
2018-06-19 11:35:17.689  INFO 25136 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@15bb6bea: startup date [Tue Jun 19 11:35:17 EDT 2018]; root of context hierarchy
2018-06-19 11:35:18.135  INFO 25136 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Starting...
2018-06-19 11:35:18.136  WARN 25136 --- [           main] com.zaxxer.hikari.util.DriverDataSource  : Registered driver with driverClassName=org.hsqldb.jdbcDriver was not found, trying direct instantiation.
2018-06-19 11:35:18.282  INFO 25136 --- [           main] com.zaxxer.hikari.pool.PoolBase          : HikariPool-1 - Driver does not support get/set network timeout for connections. (feature not supported)
2018-06-19 11:35:18.284  INFO 25136 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.
2018-06-19 11:35:18.293  INFO 25136 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executing SQL script from URL [file:/C:/eclipse-workspace/gs-batch-processing-master/complete/target/classes/schema-all.sql]
2018-06-19 11:35:18.297  INFO 25136 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executed SQL script from URL [file:/C:/eclipse-workspace/gs-batch-processing-master/complete/target/classes/schema-all.sql] in 4 ms.
2018-06-19 11:35:18.518  INFO 25136 --- [           main] org.mongodb.driver.cluster               : Cluster created with settings {hosts=[localhost:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2018-06-19 11:35:18.552  INFO 25136 --- [localhost:27017] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:1, serverValue:140}] to localhost:27017
2018-06-19 11:35:18.554  INFO 25136 --- [localhost:27017] org.mongodb.driver.cluster               : Monitor thread successfully connected to server with description ServerDescription{address=localhost:27017, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 0]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=1438717}
2018-06-19 11:35:18.723  INFO 25136 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: HSQL
2018-06-19 11:35:18.770  INFO 25136 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : No TaskExecutor has been set, defaulting to synchronous executor.
2018-06-19 11:35:18.778  INFO 25136 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executing SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql]
2018-06-19 11:35:18.781  INFO 25136 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executed SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql] in 3 ms.
2018-06-19 11:35:18.870  INFO 25136 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
2018-06-19 11:35:18.871  INFO 25136 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Bean with name 'dataSource' has been autodetected for JMX exposure
2018-06-19 11:35:18.873  INFO 25136 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located MBean 'dataSource': registering with JMX server as MBean [com.zaxxer.hikari:name=dataSource,type=HikariDataSource]
2018-06-19 11:35:18.880  INFO 25136 --- [           main] hello.Application                        : Started Application in 1.357 seconds (JVM running for 2.284)
2018-06-19 11:35:18.881  INFO 25136 --- [           main] o.s.b.a.b.JobLauncherCommandLineRunner   : Running default command line with: []
2018-06-19 11:35:18.908  INFO 25136 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=exportUserJob]] launched with the following parameters: [{run.id=1}]
===== listening for job - mongoReader - fileWriter ====
2018-06-19 11:35:18.917  INFO 25136 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step2]
2018-06-19 11:35:18.995  INFO 25136 --- [           main] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:2, serverValue:141}] to localhost:27017
2018-06-19 11:35:19.022  INFO 25136 --- [           main] bbye.PersonDocProcessor                  : Converting (firstName: ALICE, lastName: WONDERLAND) into (firstName: alice, lastName: wonderland)
2018-06-19 11:35:19.022  INFO 25136 --- [           main] bbye.PersonDocProcessor                  : Converting (firstName: FIRSTNAME, lastName: LASTNAME) into (firstName: firstname, lastName: lastname)
2018-06-19 11:35:19.022  INFO 25136 --- [           main] bbye.PersonDocProcessor                  : Converting (firstName: JANE, lastName: DOE) into (firstName: jane, lastName: doe)
2018-06-19 11:35:19.022  INFO 25136 --- [           main] bbye.PersonDocProcessor                  : Converting (firstName: JOHN, lastName: DOE) into (firstName: john, lastName: doe)
2018-06-19 11:35:19.022  INFO 25136 --- [           main] bbye.PersonDocProcessor                  : Converting (firstName: MARK, lastName: WINN) into (firstName: mark, lastName: winn)
==== file write job completed =====
2018-06-19 11:35:19.031  INFO 25136 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=exportUserJob]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]

2018-06-19 11:35:19.032  INFO 25136 --- [       Thread-2] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@15bb6bea: startup date [Tue Jun 19 11:35:17 EDT 2018]; root of context hierarchy
2018-06-19 11:35:19.033  INFO 25136 --- [       Thread-2] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown
2018-06-19 11:35:19.034  INFO 25136 --- [       Thread-2] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans
2018-06-19 11:35:19.035  INFO 25136 --- [       Thread-2] org.mongodb.driver.connection            : Closed connection [connectionId{localValue:2, serverValue:141}] to localhost:27017 because the pool has been closed.
2018-06-19 11:35:19.036  INFO 25136 --- [       Thread-2] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown initiated...
2018-06-19 11:35:19.037  INFO 25136 --- [       Thread-2] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown completed.

Answer 1

我认为我们应该使用 FileSystemResource 而不是 ClassPathResource。您能尝试告诉我们吗？