Spring批处理CSV处理

总览

我们将讨论的主题包括使用Spring Batch进行批处理的基本概念，以及如何将数据从CSV导入数据库。

0 – Spring Batch CSV处理示例应用程序

我们正在构建一个应用程序，该应用程序演示Spring Batch处理CSV文件的基础。我们的演示应用程序将允许我们处理CSV文件，其中包含数百条日本动漫标题的记录。

0.1 – CSV

我已经从这个Github存储库中下载了将要使用的CSV文件，它提供了相当全面的动漫列表。

这是在Microsoft Excel中打开的CSV的屏幕截图

查看并从 Github 下载代码

1 –项目结构

2 –项目依赖性

除了典型的Spring Boot依赖关系外，我们还包括spring-boot-starter-batch（这是对Spring Batch的依赖，顾名思义）和hsqldb（用于内存数据库）。我们还包括ToStringBuilder的commons-lang3。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.michaelcgood</groupId><artifactId>michaelcgood-spring-batch-csv</artifactId><version>0.0.1</version><packaging>jar</packaging><name>michaelcgood-spring-batch-csv</name><description>Michael C  Good - Spring Batch CSV Example Application</description><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.7.RELEASE</version><relativePath /> <!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-batch</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-jpa</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.hsqldb</groupId><artifactId>hsqldb</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 --><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.6</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

3 –模型

这是对动漫领域进行建模的POJO。字段是：

ID。为了简单起见，我们将ID视为字符串。但是，可以将其更改为其他数据类型，例如Integer或Long。
标题。这是动画的标题，适合作为String。
描述。这是动漫的描述，比标题长，也可以视为字符串。

需要注意的是我们的三个字段的类构造函数：public AnimeDTO（字符串id，字符串标题，字符串描述）。这将在我们的应用程序中使用。同样，像往常一样，我们需要创建一个没有参数的默认构造函数，否则Java会抛出错误。

package com.michaelcgood;import org.apache.commons.lang3.builder.ToStringBuilder;
/*** Contains the information of a single anime** @author Michael C Good michaelcgood.com*/public class AnimeDTO {public String getId() {return id;}public void setId(String id) {this.id = id;}public String getTitle() {return title;}public void setTitle(String title) {this.title = title;}public String getDescription() {return description;}public void setDescription(String description) {this.description = description;}private String id;private String title;private String description;public AnimeDTO(){}public AnimeDTO(String id, String title, String description){this.id = id;this.title = title;this.description = title;}@Overridepublic String toString() {return new ToStringBuilder(this).append("id", this.id).append("title", this.title).append("description", this.description).toString();}}

4 – CSV文件到数据库配置

该类中发生了很多事情，并且不是一次编写的，因此我们将逐步学习代码。访问Github以查看完整的代码。

4.1 –读者

如Spring Batch文档所述，FlatFileIteamReader将“从平面文件中读取数据行，这些文件通常描述记录的数据字段由文件中的固定位置定义或由某些特殊字符（例如，逗号）分隔”。

我们正在处理CSV，因此，当然用逗号分隔数据，这使其非常适合与我们的文件一起使用。

@Beanpublic FlatFileItemReader<AnimeDTO> csvAnimeReader(){FlatFileItemReader<AnimeDTO> reader = new FlatFileItemReader<AnimeDTO>();reader.setResource(new ClassPathResource("animescsv.csv"));reader.setLineMapper(new DefaultLineMapper<AnimeDTO>() {{setLineTokenizer(new DelimitedLineTokenizer() {{setNames(new String[] { "id", "title", "description" });}});setFieldSetMapper(new BeanWrapperFieldSetMapper<AnimeDTO>() {{setTargetType(AnimeDTO.class);}});}});return reader;}

重要事项：

- FlatFileItemReader使用模型进行参数化。

4.2 –处理器

如果要在将数据写入数据库之前对其进行转换，则需要一个ItemProcessor。我们的代码实际上并没有应用任何业务逻辑来转换数据，但是我们允许这种能力。

4.2.1 – CsvFileToDatabaseConfig.Java中的处理器

csvAnimeProcessor返回AnimeProcessor对象的新实例，我们将在下面进行检查。

@BeanItemProcessor<AnimeDTO, AnimeDTO> csvAnimeProcessor() {return new AnimeProcessor();}

4.2.2 – AnimeProcessor.Java

如果我们想在写入数据库之前应用业务逻辑，则可以在写入数据库之前操纵字符串。例如，您可以在getTitle之后添加toUpperCase（）以使标题大写，然后再写入数据库。但是，我决定不对此示例处理器执行此操作或不应用任何其他业务逻辑，因此未进行任何操作。该处理器仅在此处进行演示。

package com.michaelcgood;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import org.springframework.batch.item.ItemProcessor;public class AnimeProcessor implements ItemProcessor<AnimeDTO, AnimeDTO> {private static final Logger log = LoggerFactory.getLogger(AnimeProcessor.class);@Overridepublic AnimeDTO process(final AnimeDTO AnimeDTO) throws Exception {final String id = AnimeDTO.getId();final String title = AnimeDTO.getTitle();final String description = AnimeDTO.getDescription();final AnimeDTO transformedAnimeDTO = new AnimeDTO(id, title, description);log.info("Converting (" + AnimeDTO + ") into (" + transformedAnimeDTO + ")");return transformedAnimeDTO;}}

4.3 –作家

csvAnimeWriter方法负责将值实际写入我们的数据库。我们的数据库是内存中的HSQLDB，但是此应用程序使我们可以轻松地将一个数据库换成另一个数据库。 dataSource是自动连线的。

@Beanpublic JdbcBatchItemWriter<AnimeDTO> csvAnimeWriter() {JdbcBatchItemWriter<AnimeDTO> excelAnimeWriter = new JdbcBatchItemWriter<AnimeDTO>();excelAnimeWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<AnimeDTO>());excelAnimeWriter.setSql("INSERT INTO animes (id, title, description) VALUES (:id, :title, :description)");excelAnimeWriter.setDataSource(dataSource);return excelAnimeWriter;}

4.4 –步骤

步骤是一个域对象，它包含批处理作业的独立顺序阶段，并包含定义和控制实际批处理所需的所有信息。

现在，我们已经为数据创建了读取器和处理器，我们需要编写数据。对于读取，我们一直在使用面向块的处理，这意味着我们一次读取了一个数据。面向块的处理还包括在事务边界内创建将被写出的“块”。对于面向块的处理，您可以设置提交间隔，一旦读取的项目数等于已设置的提交间隔，就可以通过ItemWriter写入整个块，并提交事务。我们将块间隔大小设置为1。

我建议阅读有关面向块处理的Spring Batch文档。

然后，读取器，处理器和写入器调用我们编写的方法。

@Beanpublic Step csvFileToDatabaseStep() {return stepBuilderFactory.get("csvFileToDatabaseStep").<AnimeDTO, AnimeDTO>chunk(1).reader(csvAnimeReader()).processor(csvAnimeProcessor()).writer(csvAnimeWriter()).build();}

4.5 –工作

作业由步骤组成。我们将参数传递到下面的Job中，因为我们想跟踪Job的完成情况。

@BeanJob csvFileToDatabaseJob(JobCompletionNotificationListener listener) {return jobBuilderFactory.get("csvFileToDatabaseJob").incrementer(new RunIdIncrementer()).listener(listener).flow(csvFileToDatabaseStep()).end().build();}

5 –作业完成通知监听器

下面的类自动连接JdbcTemplate，因为我们已经设置了dataSource并且我们想轻松地进行查询。我们查询的结果是AnimeDTO对象的列表。对于返回的每个对象，我们将在控制台中创建一条消息，以显示该项目已被写入数据库。

package com.michaelcgood;import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Component;@Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);private final JdbcTemplate jdbcTemplate;@Autowiredpublic JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {this.jdbcTemplate = jdbcTemplate;}@Overridepublic void afterJob(JobExecution jobExecution) {if(jobExecution.getStatus() == BatchStatus.COMPLETED) {log.info("============ JOB FINISHED ============ Verifying the results....\n");List<AnimeDTO> results = jdbcTemplate.query("SELECT id, title, description FROM animes", new RowMapper<AnimeDTO>() {@Overridepublic AnimeDTO mapRow(ResultSet rs, int row) throws SQLException {return new AnimeDTO(rs.getString(1), rs.getString(2), rs.getString(3));}});for (AnimeDTO AnimeDTO : results) {log.info("Discovered <" + AnimeDTO + "> in the database.");}}}}

6 – SQL

我们需要为我们的数据库创建一个模式。如前所述，我们已将所有字段都设置为字符串，以便于使用，因此我们将其数据类型设置为VARCHAR。

DROP TABLE animes IF EXISTS;
CREATE TABLE animes  (id VARCHAR(10),title VARCHAR(400),description VARCHAR(999)
);

6 –主

这是带有main（）的标准类。如Spring文档所述， @SpringBootApplication是一个方便注释，其中包括@ Configuration ， @EnableAutoConfiguration ， @ EnableWebMvc和@ComponentScan 。

package com.michaelcgood;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class SpringBatchCsvApplication {public static void main(String[] args) {SpringApplication.run(SpringBatchCsvApplication.class, args);}
}

7 –演示

7.1 –转换

FieldSet通过处理器输入，“ Converting”被打印到控制台。

7.2 –在数据库中发现新项目

当Spring Batch Job完成时，我们选择所有记录并将它们分别打印到控制台。

7.3 –批处理完成

批处理完成后，这就是打印到控制台的内容。

Job: [FlowJob: [name=csvFileToDatabaseJob]] completed with the following parameters: [{run.id=1, -spring.output.ansi.enabled=always}] and the following status: [COMPLETED]
Started SpringBatchCsvApplication in 36.0 seconds (JVM running for 46.616)