iSharkFly-Docs
/
java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BAEL-243 Spring Batch using Partitioner (#2419) 

* BAEL-1004 changes in variable names

* BAEL-243 Spring Batch using Partitioner
This commit is contained in:
Parth Karia 2017-08-11 20:39:36 +05:30 committed by maibin
parent 69d8c10751
commit 127dd7323d
13 changed files with 399 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
spring-batch/src/main/java/org/baeldung/spring_batch_intro/partitioner/CustomMultiResourcePartitioner.java Normal file
View File

@ -0,0 +1,77 @@
/*
* Copyright 2006-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.baeldung.spring_batch_intro.partitioner;
import java.util.HashMap;
import java.util.Map;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
public class CustomMultiResourcePartitioner implements Partitioner {
private static final String DEFAULT_KEY_NAME = "fileName";
private static final String PARTITION_KEY = "partition";
private Resource[] resources = new Resource[0];
private String keyName = DEFAULT_KEY_NAME;
/**
* The resources to assign to each partition. In Spring configuration you
* can use a pattern to select multiple resources.
* @param resources the resources to use
*/
public void setResources(Resource[] resources) {
this.resources = resources;
}
/**
* The name of the key for the file name in each {@link ExecutionContext}.
* Defaults to "fileName".
* @param keyName the value of the key
*/
public void setKeyName(String keyName) {
this.keyName = keyName;
}
/**
* Assign the filename of each of the injected resources to an
* {@link ExecutionContext}.
*
* @see Partitioner#partition(int)
*/
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
Map<String, ExecutionContext> map = new HashMap<String, ExecutionContext>(gridSize);
int i = 0, k = 1;
for (Resource resource : resources) {
ExecutionContext context = new ExecutionContext();
Assert.state(resource.exists(), "Resource does not exist: " + resource);
context.putString(keyName, resource.getFilename());
context.putString("opFileName", "output" + k++ + ".xml");
map.put(PARTITION_KEY + i, context);
i++;
}
return map;
}
}

169
spring-batch/src/main/java/org/baeldung/spring_batch_intro/partitioner/SpringbatchPartitionConfig.java Normal file
View File

@ -0,0 +1,169 @@
package org.baeldung.spring_batch_intro.partitioner;
import java.io.IOException;
import java.net.MalformedURLException;
import java.text.ParseException;
import javax.sql.DataSource;
import org.baeldung.spring_batch_intro.model.Transaction;
import org.baeldung.spring_batch_intro.service.RecordFieldSetMapper;
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.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.xml.StaxEventItemWriter;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
import org.springframework.oxm.Marshaller;
import org.springframework.oxm.jaxb.Jaxb2Marshaller;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
@Configuration
@EnableBatchProcessing
public class SpringbatchPartitionConfig {
@Autowired
ResourcePatternResolver resoursePatternResolver;
@Autowired
private JobBuilderFactory jobs;
@Autowired
private StepBuilderFactory steps;
@Bean(name = "partitionerJob")
public Job partitionerJob() throws UnexpectedInputException, MalformedURLException, ParseException {
return jobs.get("partitionerJob")
.start(partitionStep())
.build();
}
@Bean
public Step partitionStep() throws UnexpectedInputException, MalformedURLException, ParseException {
return steps.get("partitionStep")
.partitioner("slaveStep", partitioner())
.step(slaveStep())
.taskExecutor(taskExecutor())
.build();
}
@Bean
public Step slaveStep() throws UnexpectedInputException, MalformedURLException, ParseException {
return steps.get("slaveStep")
.<Transaction, Transaction> chunk(1)
.reader(itemReader(null))
.writer(itemWriter(marshaller(), null))
.build();
}
@Bean
public CustomMultiResourcePartitioner partitioner() {
CustomMultiResourcePartitioner partitioner = new CustomMultiResourcePartitioner();
Resource[] resources;
try {
resources = resoursePatternResolver.getResources("file:src/main/resources/input/partitioner/*.csv");
} catch (IOException e) {
throw new RuntimeException("I/O problems when resolving the input file pattern.", e);
}
partitioner.setResources(resources);
return partitioner;
}
@Bean
@StepScope
public FlatFileItemReader<Transaction> itemReader(@Value("#{stepExecutionContext[fileName]}") String filename) throws UnexpectedInputException, ParseException {
FlatFileItemReader<Transaction> reader = new FlatFileItemReader<Transaction>();
DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
String[] tokens = { "username", "userid", "transactiondate", "amount" };
tokenizer.setNames(tokens);
reader.setResource(new ClassPathResource("input/partitioner/" + filename));
DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<Transaction>();
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper(new RecordFieldSetMapper());
reader.setLinesToSkip(1);
reader.setLineMapper(lineMapper);
return reader;
}
@Bean(destroyMethod = "")
@StepScope
public StaxEventItemWriter<Transaction> itemWriter(Marshaller marshaller, @Value("#{stepExecutionContext[opFileName]}") String filename) throws MalformedURLException {
StaxEventItemWriter<Transaction> itemWriter = new StaxEventItemWriter<>();
itemWriter.setMarshaller(marshaller);
itemWriter.setRootTagName("transactionRecord");
itemWriter.setResource(new FileSystemResource("src/main/resources/output/" + filename));
return itemWriter;
}
@Bean
public Marshaller marshaller() {
Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
marshaller.setClassesToBeBound(new Class[] { Transaction.class });
return marshaller;
}
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setMaxPoolSize(5);
taskExecutor.setCorePoolSize(5);
taskExecutor.setQueueCapacity(5);
taskExecutor.afterPropertiesSet();
return taskExecutor;
}
private JobRepository getJobRepository() throws Exception {
JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
factory.setDataSource(dataSource());
factory.setTransactionManager(getTransactionManager());
// JobRepositoryFactoryBean's methods Throws Generic Exception,
// it would have been better to have a specific one
factory.afterPropertiesSet();
return (JobRepository) factory.getObject();
}
private DataSource dataSource() {
EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
EmbeddedDatabase db = builder.setType(EmbeddedDatabaseType.HSQL)
.addScript("classpath:org/springframework/batch/core/schema-drop-h2.sql")
.addScript("classpath:org/springframework/batch/core/schema-h2.sql")
.build();
return db;
}
private PlatformTransactionManager getTransactionManager() {
return new ResourcelessTransactionManager();
}
public JobLauncher getJobLauncher() throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
// SimpleJobLauncher's methods Throws Generic Exception,
// it would have been better to have a specific one
jobLauncher.setJobRepository(getJobRepository());
jobLauncher.afterPropertiesSet();
return jobLauncher;
}
}

28
spring-batch/src/main/java/org/baeldung/spring_batch_intro/partitioner/SpringbatchPartitionerApp.java Normal file
View File

@ -0,0 +1,28 @@
package org.baeldung.spring_batch_intro.partitioner;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
public class SpringbatchPartitionerApp {
public static void main(final String[] args) {
// Spring Java config
final AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(SpringbatchPartitionConfig.class);
context.refresh();
final JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
final Job job = (Job) context.getBean("partitionerJob");
System.out.println("Starting the batch job");
try {
final JobExecution execution = jobLauncher.run(job, new JobParameters());
System.out.println("Job Status : " + execution.getStatus());
System.out.println("Job succeeded");
} catch (final Exception e) {
e.printStackTrace();
System.out.println("Job failed");
}
}
}

4
spring-batch/src/main/resources/input/partitioner/record1.csv Normal file
View File

@ -0,0 +1,4 @@
username, user_id, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
1 username user_id transaction_date transaction_amount
2 devendra 1234 31/10/2015 10000
3 john 2134 3/12/2015 12321
4 robin 2134 2/02/2015 23411

4
spring-batch/src/main/resources/input/partitioner/record2.csv Normal file
View File

@ -0,0 +1,4 @@
username, user_id, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
1 username user_id transaction_date transaction_amount
2 devendra 1234 31/10/2015 10000
3 john 2134 3/12/2015 12321
4 robin 2134 2/02/2015 23411

4
spring-batch/src/main/resources/input/partitioner/record3.csv Normal file
View File

@ -0,0 +1,4 @@
username, user_id, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
1 username user_id transaction_date transaction_amount
2 devendra 1234 31/10/2015 10000
3 john 2134 3/12/2015 12321
4 robin 2134 2/02/2015 23411

4
spring-batch/src/main/resources/input/partitioner/record4.csv Normal file
View File

@ -0,0 +1,4 @@
username, user_id, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
1 username user_id transaction_date transaction_amount
2 devendra 1234 31/10/2015 10000
3 john 2134 3/12/2015 12321
4 robin 2134 2/02/2015 23411

4
spring-batch/src/main/resources/input/partitioner/record5.csv Normal file
View File

@ -0,0 +1,4 @@
username, user_id, transaction_date, transaction_amount
devendra, 1234, 31/10/2015, 10000
john, 2134, 3/12/2015, 12321
robin, 2134, 2/02/2015, 23411
1 username user_id transaction_date transaction_amount
2 devendra 1234 31/10/2015 10000
3 john 2134 3/12/2015 12321
4 robin 2134 2/02/2015 23411

21
spring-batch/src/main/resources/output/output1.xml Normal file
View File

@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<transactionRecord>
<transactionRecord>
<amount>10000.0</amount>
<transactionDate>2015-10-31T00:00:00+05:30</transactionDate>
<userId>1234</userId>
<username>devendra</username>
</transactionRecord>
<transactionRecord>
<amount>12321.0</amount>
<transactionDate>2015-12-03T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>john</username>
</transactionRecord>
<transactionRecord>
<amount>23411.0</amount>
<transactionDate>2015-02-02T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>robin</username>
</transactionRecord>
</transactionRecord>

21
spring-batch/src/main/resources/output/output2.xml Normal file
View File

@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<transactionRecord>
<transactionRecord>
<amount>10000.0</amount>
<transactionDate>2015-10-31T00:00:00+05:30</transactionDate>
<userId>1234</userId>
<username>devendra</username>
</transactionRecord>
<transactionRecord>
<amount>12321.0</amount>
<transactionDate>2015-12-03T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>john</username>
</transactionRecord>
<transactionRecord>
<amount>23411.0</amount>
<transactionDate>2015-02-02T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>robin</username>
</transactionRecord>
</transactionRecord>

21
spring-batch/src/main/resources/output/output3.xml Normal file
View File

@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<transactionRecord>
<transactionRecord>
<amount>10000.0</amount>
<transactionDate>2015-10-31T00:00:00+05:30</transactionDate>
<userId>1234</userId>
<username>devendra</username>
</transactionRecord>
<transactionRecord>
<amount>12321.0</amount>
<transactionDate>2015-12-03T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>john</username>
</transactionRecord>
<transactionRecord>
<amount>23411.0</amount>
<transactionDate>2015-02-02T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>robin</username>
</transactionRecord>
</transactionRecord>

21
spring-batch/src/main/resources/output/output4.xml Normal file
View File

@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<transactionRecord>
<transactionRecord>
<amount>10000.0</amount>
<transactionDate>2015-10-31T00:00:00+05:30</transactionDate>
<userId>1234</userId>
<username>devendra</username>
</transactionRecord>
<transactionRecord>
<amount>12321.0</amount>
<transactionDate>2015-12-03T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>john</username>
</transactionRecord>
<transactionRecord>
<amount>23411.0</amount>
<transactionDate>2015-02-02T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>robin</username>
</transactionRecord>
</transactionRecord>

21
spring-batch/src/main/resources/output/output5.xml Normal file
View File

@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<transactionRecord>
<transactionRecord>
<amount>10000.0</amount>
<transactionDate>2015-10-31T00:00:00+05:30</transactionDate>
<userId>1234</userId>
<username>devendra</username>
</transactionRecord>
<transactionRecord>
<amount>12321.0</amount>
<transactionDate>2015-12-03T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>john</username>
</transactionRecord>
<transactionRecord>
<amount>23411.0</amount>
<transactionDate>2015-02-02T00:00:00+05:30</transactionDate>
<userId>2134</userId>
<username>robin</username>
</transactionRecord>
</transactionRecord>