Spring Batch 轻量级批处理框架实践
2023-04-18 15:00:32 时间
实践内容
从 MariaDB 一张表内读 10 万条记录,经处理后写到 MongoDB 。
具体实现
1、新建 Spring Boot 应用,依赖如下:
<!-- Web 应用 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- Web 容器 undertow -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-undertow</artifactId>
</dependency>
<!-- 日志 Log4j2 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>
<!-- MongoDB -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb</artifactId>
</dependency>
<!-- Brantch -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<!-- Mariadb 驱动 -->
<dependency>
<groupId>org.mariadb.jdbc</groupId>
<artifactId>mariadb-java-client</artifactId>
<version>2.0.2</version>
</dependency>
<!-- Lombok 代码简化 -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.14</version>
</dependency>
2、创建一张表,并生成 10 万条数据
DROP TABLE people IF EXISTS;
CREATE TABLE people (
id BIGINT IDENTITY NOT NULL PRIMARY KEY,
first_name VARCHAR(20),
last_name VARCHAR(20)
);
3、创建 Person 类
@Data
public class Person {
private Long id;
private String lastName;
private String firstName;
}
4、创建一个中间处理器 PersonItemProcessor
import org.springframework.batch.item.ItemProcessor;
@Log4j2
public class PersonItemProcessor implements ItemProcessor<Person, Person> {
@Override
public Person process(final Person person) throws Exception {
final String firstName = person.getFirstName().toUpperCase();
final String lastName = person.getLastName().toUpperCase();
final Person transformedPerson = new Person(firstName, lastName);
log.info("Converting (" + person + ") into (" + transformedPerson + ")");
return transformedPerson;
}
}
5、创建 PersonMapper
,用户数据库映射
public class PersonMapper implements RowMapper {
private static final String ID_COLUMN = "id";
private static final String NICKNAME_COLUMN = "first_name";
private static final String EMAIL_COLUMN = "last_name";
@Override
public Object mapRow(ResultSet resultSet, int i) throws SQLException {
Person user = new Person();
person.setId(resultSet.getLong(ID_COLUMN));
person.setNickname(resultSet.getString(NICKNAME_COLUMN));
person.setEmail(resultSet.getString(EMAIL_COLUMN));
return person;
}
}
6、创建任务完成的监听 JobCompletionNotificationListener
@Log4j2
@Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
@Override
public void afterJob(JobExecution jobExecution) {
if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
log.info("!!! JOB FINISHED! Time to verify the results");
}
}
}
7、构建批处理任务 BatchConfiguration
@Configuration
@EnableBatchProcessing
public class BatchConfiguration {
@Autowired
public JobBuilderFactory jobBuilderFactory;
@Autowired
public StepBuilderFactory stepBuilderFactory;
@Autowired
public DataSource dataSource;
@Autowired
public MongoTemplate mongoTemplate;
@Bean
public JdbcCursorItemReader<Person> reader(){
JdbcCursorItemReader<Person> itemReader = new JdbcCursorItemReader<Person>();
itemReader.setDataSource(dataSource);
itemReader.setSql("select id, nickname, email from people");
itemReader.setRowMapper(new PersonMapper());
return itemReader;
}
@Bean
public PersonItemProcessor processor() {
return new PersonItemProcessor();
}
@Bean
MongoItemWriter<Person> writer(){
MongoItemWriter<Person> itemWriter = new MongoItemWriter<Person>();
itemWriter.setTemplate(mongoTemplate);
itemWriter.setCollection("branch");
return itemWriter;
}
@Bean
public Step step() {
return stepBuilderFactory.get("step")
.<Person, Person> chunk(10)
.reader(reader())
.processor(processor())
.writer(writer())
.build();
}
@Bean
public Job importUserJob(JobCompletionNotificationListener listener) {
return jobBuilderFactory.get("importUserJob")
.incrementer(new RunIdIncrementer())
.listener(listener)
.flow(step())
.end()
.build();
}
}
任务处理结果
0出错,耗时 2 分钟左右,测试机 Mac
相关文章
- 编程语言是什么?
- 编程思想
- 手把手教你做一个天猫精灵(六、将QQ接入智能终端)
- AI绘画,Midjourney极简入门
- ping命令的多种玩法,以前竟然只用它来测试网速!
- 学习笔记——Django项目中的结果集、分页
- 学习笔记——Django项目中的F对象,Q对象,聚合函数,排序
- Django项目中表的查询的操作
- 【PlayWright】报错:AttributeError: 'PlaywrightContextManager' object has no attribute '_playwright'的可能解决方案
- pyqt入门
- 程序员必备的6个好习惯,成为更优秀的自己
- 在浏览器中Django项目的静态文件打不开的一个原因
- Django 使用VScode 创建工程
- 2022版最新最详细Manim开发环境搭建
- 爱心树代码实现爱心表白树源码(免费分享)html+js+css
- 撩妹/撩仔代码
- 装饰器第1天
- 基于HTML+CSS+JS实现七夕情人节表白代码【含代码】
- 可视化软件之Power BI的安装与下载
- 【SPSS】两配对样本T检验分析详细操作教程(附案例实战)