Tutoriel sur la mise en oeuvre de Spring Batch avec Spring Boot

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.2.4.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.gkemayo</groupId>
	<artifactId>datawarehouse-loader-batch</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>datawarehouse-loader-batch</name>
	<description>Spring Batch project for extracting data on files resources and load them on a DataWarehouse</description>

	<properties>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-batch</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-jdbc</artifactId>
		</dependency>

		<dependency>
			<groupId>com.h2database</groupId>
			<artifactId>h2</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-configuration-processor</artifactId>
			<optional>true</optional>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<optional>true</optional>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
			<exclusions>
				<exclusion>
					<groupId>org.junit.vintage</groupId>
					<artifactId>junit-vintage-engine</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<groupId>org.springframework.batch</groupId>
			<artifactId>spring-batch-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>

############# DataSource Config #################
spring.datasource.name=datawarehouse-db
spring.datasource.username=sa
spring.datasource.password=
spring.datasource.url= jdbc:h2:file:./src/main/resources/database/datawarehouse-db
spring.datasource.driver-class-name=org.h2.Driver
spring.datasource.sql-script-encoding= UTF-8

spring.datasource.schema = classpath:schema/datawarehouse-schema.sql

spring.h2.console.enabled=true
spring.h2.console.path=/h2-console
#http://localhost:8080/h2-console/

logging.level.root=info
logging.file.max-size=10MB
logging.file=./logs/batch-datawarehouse-loader.log

path.to.the.work.dir=../../datawarehouse-reports-repository

package com.gkemayo.batch;

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.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;

import com.gkemayo.batch.batchutils.BatchJobListener;
import com.gkemayo.batch.batchutils.BatchStepSkipper;
import com.gkemayo.batch.dto.ConvertedInputData;
import com.gkemayo.batch.dto.InputData;
import com.gkemayo.batch.processor.BatchProcessor;
import com.gkemayo.batch.reader.BatchReader;
import com.gkemayo.batch.writer.BatchWriter;

@SpringBootApplication
@EnableBatchProcessing
public class BatchDatawarehouseLoaderApplication {

	public static void main(String[] args) {
		SpringApplication.run(BatchDatawarehouseLoaderApplication.class, args);
	}
	
	@Value("${path.to.the.work.dir}")
	private String workDirPath ;
	
	@Autowired
	private DataSource dataSource;
	
	@Autowired
	PlatformTransactionManager transactionManager;
	
	@Bean
	public JobRepository jobRepositoryObj() throws Exception {
		JobRepositoryFactoryBean jobRepoFactory = new JobRepositoryFactoryBean();
		jobRepoFactory.setTransactionManager(transactionManager);
		jobRepoFactory.setDataSource(dataSource);
		return jobRepoFactory.getObject();
	}
	
	@Autowired
	JobBuilderFactory jobBuilderFactory;
	
	@Autowired
	StepBuilderFactory stepBuilderFactory;
	
	@Bean
	public BatchReader batchReader() {
		return new BatchReader(workDirPath);
	}
	
	@Bean
	public BatchProcessor batchProcessor() {
		return new BatchProcessor();
	}
	
	@Bean
	public BatchWriter batchWriter() {
		return new BatchWriter();
	}
	
	@Bean
	public BatchJobListener batchJobListener() {
		return new BatchJobListener();
	}
	
	@Bean
	public BatchStepSkipper batchStepSkipper() {
		return new BatchStepSkipper();
	}
	
	@Bean
	public Step batchStep() {
		return stepBuilderFactory.get("stepDatawarehouseLoader").transactionManager(transactionManager)
				.<InputData, ConvertedInputData>chunk(1).reader(batchReader()).processor(batchProcessor())
				.writer(batchWriter()).faultTolerant().skipPolicy(batchStepSkipper()).build();
	}

	@Bean
	public Job jobStep() throws Exception {
		return jobBuilderFactory.get("jobDatawarehouseLoader").repository(jobRepositoryObj()).incrementer(new RunIdIncrementer()).listener(batchJobListener())
				.flow(batchStep()).end().build();
	}

}

package com.gkemayo.batch.entitiesrepository;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.jdbc.repository.config.AbstractJdbcConfiguration;
import org.springframework.data.jdbc.repository.config.EnableJdbcRepositories;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;

@Configuration
@EnableJdbcRepositories
public class BatchRepositoryConfig extends AbstractJdbcConfiguration {
	
	@Autowired
	private DataSource dataSource;
	
	@Bean
    NamedParameterJdbcOperations operations() {
        return new NamedParameterJdbcTemplate(dataSource);
    }
   
}

package com.gkemayo.batch.reader;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.MultiResourceItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.core.io.FileSystemResource;
import org.springframework.validation.BindException;

import com.gkemayo.batch.dto.InputData;

import lombok.extern.slf4j.Slf4j;

/**
 * Spring batch reader : read all the input data files
 * 
 * @author Georges Kemayo
 *
 */
@Slf4j
public class BatchReader extends MultiResourceItemReader<InputData>{
	
	Logger log = LoggerFactory.getLogger(BatchReader.class);
	
	public BatchReader(String workDirPath) {
		log.info("Batch reader starting to read input data in repository : " + workDirPath);
		this.setResources(getInputResources(workDirPath));
		this.setDelegate(readOneFile());
	}
	
	/**
	 * Get all the resources (files) to be readen by this spring batch reader.
	 * 
	 * @param workDirPath
	 * @return
	 */
	private FileSystemResource[] getInputResources(String workDirPath) {

		File inputDir = new File(workDirPath);
		File[] files = inputDir.listFiles();
		List<File> filesList = Arrays.asList(files);
		
		List<FileSystemResource> inputResources = filesList.stream().filter(file -> file != null && file.isFile())
                                                                         .peek(file -> log.info("Reading file : " + file.getAbsolutePath()))
                                                                           .map(file -> new FileSystemResource(file))
                                                                            .collect(Collectors.toList());
																		   
		return inputResources.toArray(new FileSystemResource[inputResources.size()]);
	}
	
	/**
	 * Set and return a FlatFileItemReader to read one resource.
	 * 
	 * @return
	 */
	private FlatFileItemReader<InputData> readOneFile() {
		
		FlatFileItemReader<InputData> resourceReader = new FlatFileItemReader<InputData>();
		
		//skip the first line which is the file header
		resourceReader.setLinesToSkip(1);
		
		resourceReader.setLineMapper(new DefaultLineMapper<InputData>() {
			
			private DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer(";");
			private FieldSetMapper<InputData> fieldSetMapper = new BeanWrapperFieldSetMapper<InputData>() {
				@Override
		        public InputData mapFieldSet(FieldSet pFielSet) throws BindException {
					InputData inputData = new InputData();
					inputData.setProductName(pFielSet.readString("productName"));
					inputData.setProductEanCode(pFielSet.readString("productEanCode"));
					inputData.setProductType(pFielSet.readString("productType"));
					inputData.setProductAmount(pFielSet.readDouble("productAmount"));
					inputData.setProductQuantity(pFielSet.readInt("productQuantity"));
					inputData.setSupplierName(pFielSet.readString("supplierName"));
					inputData.setSupplierAddress(pFielSet.readString("supplierAddress"));
					inputData.setPurchaserEmail(pFielSet.readString("purchaserEmail"));
					inputData.setPurchaserFirstName(pFielSet.readString("purchaserFirstName"));
					inputData.setPurchaserLastName(pFielSet.readString("purchaserLastName"));
					inputData.setTransactionDate(pFielSet.readDate("transactionDate"));
					return inputData;

		        }
			};
				
			@Override
			public InputData mapLine(String pLine, int pLineNumber) throws Exception {
				
				lineTokenizer.setNames(new String[]{"productName", "productEanCode", "productType", "productQuantity", 
                                                    "productAmount", "supplierName", "supplierAddress", "purchaserFirstName", 
                                                    "purchaserLastName", "purchaserEmail", "transactionDate"});
				return fieldSetMapper.mapFieldSet(lineTokenizer.tokenize(pLine));
			}

		});
		
		return resourceReader;
	}
	
}

package com.gkemayo.batch.dto;

import java.util.Date;

import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.ToString;

@Setter
@Getter
@ToString
@EqualsAndHashCode
@RequiredArgsConstructor
public class InputData {
	
	private String productName;
	
	private String productType;
	
	private String productEanCode;
	
	private Double productAmount;
	
	private Integer productQuantity;
	
	private String purchaserFirstName;

	private String purchaserLastName;
	
	private String purchaserEmail;
	
	private String supplierName;
	
	private String supplierAddress;
	
	private Date transactionDate;
}

package com.gkemayo.batch.processor;

import org.springframework.batch.item.ItemProcessor;
import org.springframework.beans.factory.annotation.Autowired;

import com.gkemayo.batch.dto.ConvertedInputData;
import com.gkemayo.batch.dto.InputData;
import com.gkemayo.batch.entities.Order;
import com.gkemayo.batch.entities.Product;
import com.gkemayo.batch.entities.PurchaseDate;
import com.gkemayo.batch.entities.Purchaser;
import com.gkemayo.batch.entities.Supplier;
import com.gkemayo.batch.entitiesrepository.ProductRepository;
import com.gkemayo.batch.entitiesrepository.PurchaseDateRepository;
import com.gkemayo.batch.entitiesrepository.PurchaserRepository;
import com.gkemayo.batch.entitiesrepository.SupplierRepository;

public class BatchProcessor implements ItemProcessor<InputData, ConvertedInputData> {
	
	@Autowired
	private SupplierRepository supplierRepository;
	
	@Autowired
	private PurchaserRepository purchaserRepository;
	
	@Autowired
	private ProductRepository productRepository;
	
	@Autowired
	private PurchaseDateRepository purchaseDateRepository;

	@Override
	public ConvertedInputData process(InputData item) throws Exception {
		
		ConvertedInputData convertedInputData = new ConvertedInputData();
		Order order = new Order();
		
		Supplier supplier = supplierRepository.findByName(item.getSupplierName());
		Purchaser purchaser = purchaserRepository.findByEmail(item.getPurchaserEmail());
		Product product = productRepository.findByEanCode(item.getProductEanCode());
		PurchaseDate purchaseDate = purchaseDateRepository.findByDate(item.getTransactionDate());
		
		if(supplier == null) {
			supplier = Supplier.of(null, item.getSupplierName(), item.getSupplierAddress());
			 convertedInputData.setSupplier(supplier);
		}else {
			order.setSupplierId(supplier.getId());
		}
		
		if(purchaser == null) {
			purchaser = Purchaser.of(null, item.getPurchaserFirstName(), item.getPurchaserLastName(), item.getPurchaserEmail());
			convertedInputData.setPurchaser(purchaser);
		}else {
			order.setPurchaserId(purchaser.getId());
		}
		
		if(product == null) {
			product = Product.of(null, item.getProductName(), item.getProductType(), item.getProductEanCode());
			convertedInputData.setProduct(product);
			order.setQuantity(item.getProductQuantity());
			order.setAmount(item.getProductAmount()*item.getProductQuantity());
		}else {
			order.setProductId(product.getId());
			order.setQuantity(item.getProductQuantity());
			order.setAmount(item.getProductAmount()*item.getProductQuantity());
		}
		
		if(purchaseDate == null) {
			purchaseDate = PurchaseDate.of(null, item.getTransactionDate());
			convertedInputData.setPurchaseDate(purchaseDate);
		}else {
			order.setDateId(purchaseDate.getId());
		}
		
		convertedInputData.setOrder(order);
		
		return convertedInputData;
	}

}

package com.gkemayo.batch.dto;

import com.gkemayo.batch.entities.Supplier;
import com.gkemayo.batch.entities.Order;
import com.gkemayo.batch.entities.Product;
import com.gkemayo.batch.entities.Purchaser;
import com.gkemayo.batch.entities.PurchaseDate;

import lombok.Data;

@Data
public class ConvertedInputData {
	
	private Supplier supplier;
	
	private Purchaser purchaser;
	
	private Product product;
	
	private PurchaseDate purchaseDate;
	
	private Order order;
}

package com.gkemayo.batch.writer;

import java.util.List;

import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;

import com.gkemayo.batch.dto.ConvertedInputData;
import com.gkemayo.batch.entities.Supplier;
import com.gkemayo.batch.entities.Product;
import com.gkemayo.batch.entities.Purchaser;
import com.gkemayo.batch.entities.PurchaseDate;
import com.gkemayo.batch.entitiesrepository.SupplierRepository;
import com.gkemayo.batch.entitiesrepository.OrderRepository;
import com.gkemayo.batch.entitiesrepository.ProductRepository;
import com.gkemayo.batch.entitiesrepository.PurchaserRepository;
import com.gkemayo.batch.entitiesrepository.PurchaseDateRepository;

public class BatchWriter implements ItemWriter<ConvertedInputData> {
	
	@Autowired
	private SupplierRepository supplierRepository;
	
	@Autowired
	private PurchaserRepository purchaserRepository;
	
	@Autowired
	private ProductRepository productRepository;
	
	@Autowired
	private PurchaseDateRepository purchaseDateRepository;
	
	@Autowired
	private OrderRepository commandRepository;

	@Override
	public void write(List<? extends ConvertedInputData> items) throws Exception {
		items.stream().forEach(item -> {
			Supplier supplier = null;
			Purchaser purchaser = null;
			Product product = null;
			PurchaseDate purchaseDate = null;
			if(item.getOrder().getSupplierId() == null) {
				supplier = supplierRepository.save(item.getSupplier());
				item.getOrder().setSupplierId(supplier.getId());
			}
			if(item.getOrder().getPurchaserId() == null) {
				purchaser = purchaserRepository.save(item.getPurchaser());
				item.getOrder().setPurchaserId(purchaser.getId());
			}
			if(item.getOrder().getProductId() == null) {
				product = productRepository.save(item.getProduct());
				item.getOrder().setProductId(product.getId());
			}
			if(item.getOrder().getDateId() == null) {
				purchaseDate = purchaseDateRepository.save(item.getPurchaseDate());
				item.getOrder().setDateId(purchaseDate.getId());
			}
			commandRepository.save(item.getOrder());
		});
	}
	
}

package com.gkemayo.batch.batchutils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.file.FlatFileParseException;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class BatchStepSkipper implements SkipPolicy {
	
	Logger log = LoggerFactory.getLogger(BatchStepSkipper.class);
	
	private static Integer SKIP_THRESHOLD = 3;

	@Override
	public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
		if(t instanceof FlatFileParseException && skipCount < SKIP_THRESHOLD) {
			log.info(((FlatFileParseException) t).getLineNumber() + ": " +((FlatFileParseException) t).getInput() + " -- " + ((FlatFileParseException) t).getMessage());
			return true;
		}
		return false;
	}
	
	

}

package com.gkemayo.batch.batchutils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class BatchJobListener implements JobExecutionListener {
	
	Logger log = LoggerFactory.getLogger(BatchJobListener.class);

	@Override
	public void beforeJob(JobExecution jobExecution) {
		log.info("Begining of job  " + jobExecution.getJobInstance().getJobName() + " at " + jobExecution.getStartTime());
	}

	@Override
	public void afterJob(JobExecution jobExecution) {
		String exitCode = jobExecution.getExitStatus().getExitCode();
		log.info("End of job  " + jobExecution.getJobInstance().getJobName() + " at " + jobExecution.getEndTime() + " with status " + exitCode);
	}

}

package com.gkemayo.batch;

import org.junit.Assert;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.batch.test.context.SpringBatchTest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
@SpringBatchTest	
@ActiveProfiles("test")
@TestPropertySource(locations = "classpath:application-test.properties")
class BatchDatawarehouseLoaderApplicationTests {
	
	@Autowired
	JobLauncherTestUtils jobLaucherTestUtils;
	
	@Test
	public void testBatchDatawarehouseLoader() throws Exception {
		JobExecution jobExecution = jobLaucherTestUtils.launchJob();
		
		Assert.assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
	}
}