Spring-batch (ItemProcessor) data processing process
Summary of Spring-batch Learning (5)
Learning goal: master ItemProcessor
The process of data processing in 1.ItemProcessor:spring-batch
2.ItemProcessor is mainly used to implement business logic, authentication, filtering, etc.
3.Spring-batch provides us with the interface ItemProcessor, which contains a method O process (I item)
4. We demonstrate with code:
Example: we read the data in the database table person_buf, remove the data whose id is odd, and convert the readout name to uppercase letters.
First, observe the data structure of the database table:
Code:
Person
Package com.dhcc.batch.batchDemo.processor;import java.util.Date;public class Person {private Integer id; private String name; private String perDesc; private Date createTime; private Date updateTime; private String sex; private Float score; private Double price; public Person () {super () } public Person (Integer id, String name, String perDesc, Date createTime, Date updateTime, String sex, Float score, Double price) {super (); this.id = id; this.name = name; this.perDesc = perDesc; this.createTime = createTime; this.updateTime = updateTime; this.sex = sex; this.score = score; this.price = price } public Integer getId () {return id;} public void setId (Integer id) {this.id = id;} public String getName () {return name;} public void setName (String name) {this.name = name;} public Date getCreateTime () {return createTime;} public String getPerDesc () {return perDesc } public void setPerDesc (String perDesc) {this.perDesc = perDesc;} public void setCreateTime (Date createTime) {this.createTime = createTime;} public Date getUpdateTime () {return updateTime;} public void setUpdateTime (Date updateTime) {this.updateTime = updateTime;} public String getSex () {return sex } public void setSex (String sex) {this.sex = sex;} public Float getScore () {return score;} public void setScore (Float score) {this.score = score;} public Double getPrice () {return price;} public void setPrice (Double price) {this.price = price @ Override public String toString () {return "Person [id=" + id + ", name=" + name + ", perDesc=" + perDesc + ", createTime=" + createTime + ", updateTime=" + updateTime + ", sex=" + sex + ", score=" + score + ", price=" + price + "]";}}
PersonLineAggregator
Package com.dhcc.batch.batchDemo.processor;import org.springframework.batch.item.file.transform.LineAggregator;import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.ObjectMapper;public class PersonLineAggregator implements LineAggregator {/ / JSON private ObjectMapper mapper=new ObjectMapper (); @ Override public String aggregate (Person person) {try {return mapper.writeValueAsString (person) } catch (JsonProcessingException e) {throw new RuntimeException ("unable to writer...", e);}
PersonRowMapper
Package com.dhcc.batch.batchDemo.processor;import java.sql.ResultSet;import java.sql.SQLException;import org.springframework.jdbc.core.RowMapper / * implement mapping each piece of data in the database to a Person object * @ author Administrator * * / public class PersonRowMapper implements RowMapper {/ * rs a result set RowNum represents the current line * / @ Override public Person mapRow (ResultSet rs, int rowNum) throws SQLException {return new Person (rs.getInt ("id"), rs.getString ("name"), rs.getString ("per_desc"), rs.getDate ("create_time"), rs.getDate ("update_time") Rs.getString ("sex"), rs.getFloat ("score"), rs.getDouble ("price")) }}
ProcessorFileApplication
Package com.dhcc.batch.batchDemo.processor;import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication@EnableBatchProcessingpublic class ProcessorFileApplication {public static void main (String [] args) {SpringApplication.run (ProcessorFileApplication.class, args);}}
ProcessorFileOutputFromDBConfiguration
Package com.dhcc.batch.batchDemo.processor;import java.io.File;import java.util.ArrayList;import java.util.HashMap;import java.util.List;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.JobBuilderFactory;import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;import org.springframework.batch.core.configuration.annotation.StepScope;import org.springframework.batch.item.ItemProcessor Import org.springframework.batch.item.database.JdbcPagingItemReader;import org.springframework.batch.item.database.Order;import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;import org.springframework.batch.item.file.FlatFileItemWriter;import org.springframework.batch.item.support.CompositeItemProcessor;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.core.io.FileSystemResource;@Configurationpublic class ProcessorFileOutputFromDBConfiguration {@ Autowired private JobBuilderFactory jobBuilderFactory @ Autowired private StepBuilderFactory stepBuilderFactory; @ Autowired private DataSource dataSource; @ Autowired private ItemProcessor fristNameUpperCaseProcessor; @ Autowired private ItemProcessor idFilterProcessor; @ Bean public Job ProcessorFileOutputFromDBJob () {return jobBuilderFactory.get ("ProcessorFileOutputFromDBJob") .start (ProcessorFileOutputFromDBStep ()) .build () } @ Bean public Step ProcessorFileOutputFromDBStep () {return stepBuilderFactory.get ("ProcessorFileOutputFromDBStep") .chunk (100) .reader (ProcessorFileOutputFromItemWriter ()) .processor (personDataProcessor ()) .writer (ProcessorFileOutputFromItemReader ()) .build ();} @ Bean @ StepScope public JdbcPagingItemReader ProcessorFileOutputFromItemWriter () {JdbcPagingItemReader reader = new JdbcPagingItemReader () Reader.setDataSource (this.dataSource); / / set the data source reader.setFetchSize (100); / / set the maximum number of entries read at one time reader.setRowMapper (new PersonRowMapper ()); / / map each piece of data in the database to the AlipaytranDo object MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider (); queryProvider.setSelectClause ("id,name,per_desc,create_time,update_time,sex,score,price") / / set the query column queryProvider.setFromClause ("from person_buf"); / / set the table to be queried Map sortKeys = new HashMap (); / / define a collection for storing the sort sortKeys.put ("id", Order.ASCENDING); / / sort queryProvider.setSortKeys (sortKeys) in ascending order; reader.setQueryProvider (queryProvider); / / set the sort return reader. @ Bean public CompositeItemProcessor personDataProcessor () {CompositeItemProcessor processor=new CompositeItemProcessor (); List listProcessor=new ArrayList (); listProcessor.add (fristNameUpperCaseProcessor); listProcessor.add (idFilterProcessor); processor.setDelegates (listProcessor); return processor;} @ Bean @ StepScope public FlatFileItemWriter ProcessorFileOutputFromItemReader () {FlatFileItemWriter writer = new FlatFileItemWriter () Try {File path = new File ("D:" + File.separator + "newPerson.json"). GetAbsoluteFile (); System.out.println ("file is create in:" + path); writer.setResource (new FileSystemResource (path)); writer.setLineAggregator (new PersonLineAggregator ()); writer.afterPropertiesSet ();} catch (Exception e) {e.printStackTrace () } return writer;}}
FristNameUpperCaseProcessor
Package com.dhcc.batch.batchDemo.processor;import org.springframework.batch.item.ItemProcessor;import org.springframework.stereotype.Component;@Componentpublic class FristNameUpperCaseProcessor implements ItemProcessor {@ Override public Person process (Person item) throws Exception {return new Person (item.getId (), item.getName (). ToUpperCase (), item.getPerDesc (), item.getCreateTime (), item.getUpdateTime (), item.getSex (), item.getScore (), item.getPrice ());}}
IdFilterProcessor
Package com.dhcc.batch.batchDemo.processor;import org.springframework.batch.item.ItemProcessor;import org.springframework.stereotype.Component;@Componentpublic class IdFilterProcessor implements ItemProcessor {@ Override public Person process (Person item) throws Exception {if (item.getId ()% 2 = = 0) {return item;} else {return null;}}
Running result:
Observe the file after the write is complete:
It can be seen that we have achieved our goal.
