详解Spring Batch 轻量级批处理框架实践

实践内容

从 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

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • 详解Spring batch 入门学习教程(附源码)

    Spring batch 是一个开源的批处理框架.执行一系列的任务. 在 spring batch 中 一个job 是由许多 step 组成的.而每一个 step  又是由 READ-PROCESS-WRITE task或者 单个 task 组成. 1. "READ-PROCESS-WRITE" 处理,根据字面意思理解就可以: READ 就是从资源文件里面读取数据,比如从xml文件,csv文件,数据库中读取数据. PROCESS 就是处理读取的数据 WRITE 就是将处理过的数据写入到

  • 浅谈Spring Batch在大型企业中的最佳实践

    在大型企业中,由于业务复杂.数据量大.数据格式不同.数据交互格式繁杂,并非所有的操作都能通过交互界面进行处理.而有一些操作需要定期读取大批量的数据,然后进行一系列的后续处理.这样的过程就是"批处理". 批处理应用通常有以下特点: 数据量大,从数万到数百万甚至上亿不等: 整个过程全部自动化,并预留一定接口进行自定义配置: 这样的应用通常是周期性运行,比如按日.周.月运行: 对数据处理的准确性要求高,并且需要容错机制.回滚机制.完善的日志监控等. 什么是Spring batch Sprin

  • springbatch的封装与使用实例详解

    Spring Batch官网介绍: A lightweight, comprehensive batch framework designed to enable the development of robust batch applications vital for the daily operations of enterprise systems.(一款轻量的.全面的批处理框架,用于开发强大的日常运营的企业级批处理应用程序.) springbatch 主要实现批量数据的处理,我对bat

  • 详解spring batch的使用和定时器Quart的使用

    spring Batch是一个基于Spring的企业级批处理框架,它通过配合定时器Quartz来轻易实现大批量的数据读取或插入,并且全程自动化,无需人员管理. 在使用spring batch之前,得对spring batch的流程有一个基本了解 每个batch它都包含了一个job,而一个job中却有可能包含多个step,整个batch中干活的是step,batch主要是用来对数据的操作,所以step就有三个操作数据的东西,一个是ItemReader用来读取数据的,一个是ItemProcessor

  • 基于Spring Batch向Elasticsearch批量导入数据示例

    1.介绍 当系统有大量数据需要从数据库导入Elasticsearch时,使用Spring Batch可以提高导入的效率.Spring Batch使用ItemReader分页读取数据,ItemWriter批量写数据.由于Spring Batch没有提供Elastisearch的ItemWriter和ItemReader,本示例中自定义一个ElasticsearchItemWriter(ElasticsearchItemReader),用于批量导入. 2.示例 2.1 pom.xml 本文使用spr

  • 详解SpringBoot和SpringBatch 使用

    什么是Spring Batch Spring Batch 是一个轻量级的.完善的批处理框架,旨在帮助企业建立健壮.高效的批处理应用.Spring Batch是Spring的一个子项目,使用Java语言并基于Spring框架为基础开发,使的已经使用 Spring 框架的开发者或者企业更容易访问和利用企业服务. Spring Batch 提供了大量可重用的组件,包括了日志.追踪.事务.任务作业统计.任务重启.跳过.重复.资源管理.对于大数据量和高性能的批处理任务,Spring Batch 同样提供了

  • Spring Batch入门教程篇

    SpringBatch介绍: SpringBatch 是一个大数据量的并行处理框架.通常用于数据的离线迁移,和数据处理,⽀持事务.并发.流程.监控.纵向和横向扩展,提供统⼀的接⼝管理和任务管理;SpringBatch是SpringSource和埃森哲为了统一业界并行处理标准为广大开发者提供方便开发的一套框架. 官方地址:github.com/spring-projects/spring-batch SpringBatch 本身提供了重试,异常处理,跳过,重启.任务处理统计,资源管理等特性,这些特

  • Spring Batch读取txt文件并写入数据库的方法教程

    项目需求 近日需要实现用户推荐相关的功能,也就是说向用户推荐他可能喜欢的东西. 我们的数据分析工程师会将用户以及用户可能喜欢的东西整理成文档给我,我只需要将数据从文档中读取出来,然后对数据进行进一步的清洗(例如去掉特殊符号,长度如果太长则截取).然后将处理后的数据存入数据库(Mysql). 所以分为三步: 读取文档获得数据 对获得的数据进行处理 更新数据库(新增或更新) 考虑到这个数据量以后会越来越大,这里没有使用 poi 来读取数据,而直接使用了 SpringBatch. 实现步骤 本文假设读

  • Spring batch批处理框架

    spring batch框架的简介 批处理任务是大多数IT项目的一个重要组成部分,批处理在业务系统中负责处理海量的数据,无须人工干预就能够自动高效的进行复杂的数据分析和处理.批处理会定期读入批量数据,经过相应的业务处理进行归档的业务操作,批处理的特征是自动执行,处理的数据量大,定时执行.将整个批处理的流程按逻辑划分可以分为读数据,处理数据和写数据. spring batch对批处理本身的特性进行了抽象,将批处理作业抽象为job和job step,将批处理的处理过程分解为数据读,数据处理和数据写.

  • spring batch 读取多个文件数据导入数据库示例

    项目的目录结构 需要读取文件的的数据格式 applicatonContext.xml的配置 <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p=

随机推荐