Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-1103: Added partitioning sample

Browse Source
This commit is contained in:
dsyer
2009-02-25 12:31:13 +00:00
parent a07a7166fb
commit 044a91eddf
10 changed files with 461 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerFactoryBean.java
View File

@@ -78,7 +78,7 @@ public class StepListenerFactoryBean implements FactoryBean, InitializingBean{
Set<Class<? extends StepListener>> listenerInterfaces = new HashSet<Class<? extends StepListener>>();
//For every entry in th emap, try and find a method by interface, name, or annotation. If the same
//For every entry in the map, try and find a method by interface, name, or annotation. If the same
for(Entry<String, String> entry : metaDataMap.entrySet()){
StepListenerMetaData metaData = StepListenerMetaData.fromPropertyName(entry.getKey());
Set<MethodInvoker> invokers = new NullIgnoringSet<MethodInvoker>();

89
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/MultiResourcePartitioner.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* Copyright 2006-2007 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.springframework.batch.core.partition.support;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
/**
* Implementation of {@link Partitioner} that locates multiple resources and
* associates their file names with execution context keys. Creates an
* {@link ExecutionContext} per resource, and labels them as
* <code>{partition0, partition1, ..., partitionN}</code>. The grid size is
* ignored.
*
* @author Dave Syer
*
*/
public class MultiResourcePartitioner 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)
*/
public Map<String, ExecutionContext> partition(int gridSize) {
Map<String, ExecutionContext> map = new HashMap<String, ExecutionContext>(gridSize);
int i = 0;
for (Resource resource : resources) {
ExecutionContext context = new ExecutionContext();
Assert.state(resource.exists(), "Resource does not exist: "+resource);
try {
context.putString(keyName, resource.getURL().toExternalForm());
}
catch (IOException e) {
throw new IllegalArgumentException("File could not be located for: "+resource, e);
}
map.put(PARTITION_KEY + i, context);
i++;
}
return map;
}
}

54
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/MultiResourcePartitionerTests.java Normal file
View File

@@ -0,0 +1,54 @@
package org.springframework.batch.core.partition.support;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.core.io.support.ResourceArrayPropertyEditor;
public class MultiResourcePartitionerTests {
private MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
@Before
public void setUp() {
ResourceArrayPropertyEditor editor = new ResourceArrayPropertyEditor();
editor.setAsText("classpath:log4j*");
partitioner.setResources((Resource[]) editor.getValue());
}
@Test(expected = IllegalStateException.class)
public void testMissingResource() {
partitioner.setResources(new Resource[] { new FileSystemResource("does-not-exist") });
partitioner.partition(0);
}
@Test
public void testPartitionSizeAndKey() {
Map<String, ExecutionContext> partition = partitioner.partition(0);
assertEquals(1, partition.size());
assertTrue(partition.containsKey("partition0"));
}
@Test
public void testReadFile() throws Exception {
Map<String, ExecutionContext> partition = partitioner.partition(0);
String url = partition.get("partition0").getString("fileName");
assertTrue(new UrlResource(url).exists());
}
@Test
public void testSetKeyName() {
partitioner.setKeyName("foo");
Map<String, ExecutionContext> partition = partitioner.partition(0);
assertTrue(partition.get("partition0").containsKey("foo"));
}
}

20
spring-batch-samples/.springBeans
View File

@@ -48,6 +48,26 @@
<config>src/main/resources/jobs/iosample/jpa.xml</config>
<config>src/main/resources/jobs/iosample/ibatis.xml</config>
<config>src/test/resources/org/springframework/batch/sample/JobOperatorFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/BeanWrapperMapperSampleJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/CustomerFilterJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/DatabaseShutdownFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/DelegatingJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/FootballJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/GracefulShutdownFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/HeaderFooterSampleFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/HibernateFailureJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/domain/trade/internal/JdbcCustomerDebitDaoTests-context.xml</config>
<config>src/main/resources/jobs/loopFlowSample.xml</config>
<config>src/test/resources/org/springframework/batch/sample/MultilineJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/MultilineOrderJobFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/ParallelJobFunctionalTests-context.xml</config>
<config>src/main/resources/jobs/partitionJob.xml</config>
<config>src/test/resources/org/springframework/batch/sample/RestartFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/RetrySampleFunctionalTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/common/StagingItemReaderTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/common/StagingItemWriterTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/sample/TradeJobFunctionalTests-context.xml</config>
</configs>
<configSets>
<configSet>

50
spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileListener.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2006-2007 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.springframework.batch.sample.common;
import org.apache.commons.io.FilenameUtils;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.item.ExecutionContext;
/**
* @author Dave Syer
*
*/
public class OutputFileListener {
private String outputKeyName = "outputFile";
private String inputKeyName = "fileName";
public void setOutputKeyName(String outputKeyName) {
this.outputKeyName = outputKeyName;
}
public void setInputKeyName(String inputKeyName) {
this.inputKeyName = inputKeyName;
}
@BeforeStep
public void CreateOutputNameFromInput(StepExecution stepExecution) {
ExecutionContext executionContext = stepExecution.getExecutionContext();
if (executionContext.containsKey(inputKeyName) && !executionContext.containsKey(outputKeyName)) {
String inputName = executionContext.getString(inputKeyName);
executionContext.putString(outputKeyName, "file:./target/output/" + FilenameUtils.getBaseName(inputName) + ".csv");
}
}
}

5
spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileNameListener.java Normal file
View File

@@ -0,0 +1,5 @@
package org.springframework.batch.sample.common;
public class OutputFileNameListener {
}

98
spring-batch-samples/src/main/resources/jobs/partitionJob.xml Normal file
View File

@@ -0,0 +1,98 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:batch="http://www.springframework.org/schema/batch" xmlns="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-2.0.xsd">
<batch:job id="partitionJob">
<batch:step id="step" ref="step1:master" />
</batch:job>
<bean name="step1:master" class="org.springframework.batch.core.partition.support.PartitionStep">
<property name="jobRepository" ref="jobRepository" />
<property name="stepExecutionSplitter">
<bean class="org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter">
<constructor-arg ref="jobRepository" />
<constructor-arg ref="step1" />
<constructor-arg>
<bean class="org.springframework.batch.core.partition.support.MultiResourcePartitioner">
<property name="resources" value="classpath:data/iosample/input/delimited*.csv" />
</bean>
</constructor-arg>
</bean>
</property>
<property name="partitionHandler">
<bean class="org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler">
<property name="taskExecutor">
<bean class="org.springframework.core.task.SyncTaskExecutor" />
</property>
<property name="step" ref="step1" />
</bean>
</property>
</bean>
<batch:step id="step1" job-repository="jobRepository" transaction-manager="transactionManager">
<batch:tasklet writer="itemWriter" reader="itemReader" processor="itemProcessor" commit-interval="5" />
<batch:listeners>
<batch:listener ref="fileNameListener" />
</batch:listeners>
</batch:step>
<bean id="fileNameListener" class="org.springframework.batch.sample.common.OutputFileListener" />
<bean id="itemReader" scope="step" autowire-candidate="false" parent="itemReaderParent">
<property name="resource" value="#{stepExecutionContext[fileName]}" />
</bean>
<bean id="inputTestReader" class="org.springframework.batch.item.file.MultiResourceItemReader">
<property name="resources" value="classpath:data/iosample/input/delimited*.csv" />
<property name="delegate" ref="testItemReader" />
</bean>
<bean id="outputTestReader" class="org.springframework.batch.item.file.MultiResourceItemReader" scope="prototype">
<property name="resources" value="file:target/output/delimited*.csv" />
<property name="delegate" ref="testItemReader" />
</bean>
<bean id="testItemReader" parent="itemReaderParent"/>
<bean id="itemReaderParent" class="org.springframework.batch.item.file.FlatFileItemReader" abstract="true">
<property name="lineMapper">
<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<property name="lineTokenizer">
<bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<property name="delimiter" value="," />
<property name="names" value="name,credit" />
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
<property name="targetType" value="org.springframework.batch.sample.domain.trade.CustomerCredit" />
</bean>
</property>
</bean>
</property>
</bean>
<bean id="itemProcessor" class="org.springframework.batch.sample.domain.trade.internal.CustomerCreditIncreaseProcessor" />
<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter" scope="step">
<property name="resource" value="#{stepExecutionContext[outputFile]}" />
<property name="lineAggregator">
<bean class="org.springframework.batch.item.file.transform.DelimitedLineAggregator">
<property name="delimiter" value="," />
<property name="fieldExtractor">
<bean class="org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor">
<property name="names" value="name,credit" />
</bean>
</property>
</bean>
</property>
</bean>
</beans>

114
spring-batch-samples/src/test/java/org/springframework/batch/sample/PartitionJobFunctionalTests.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* Copyright 2006-2007 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.springframework.batch.sample;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.sample.domain.trade.CustomerCredit;
import org.springframework.batch.sample.domain.trade.internal.CustomerCreditIncreaseProcessor;
import org.springframework.batch.test.AbstractJobTests;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration()
public class PartitionJobFunctionalTests extends AbstractJobTests {
@Autowired
@Qualifier("inputTestReader")
private ItemReader<CustomerCredit> inputReader;
/**
* Check the resulting credits correspond to inputs increased by fixed
* amount.
*/
@Test
public void testUpdateCredit() throws Exception {
assertTrue("Define a prototype bean called 'outputTestReader' to check the output", getApplicationContext()
.containsBeanDefinition("outputTestReader"));
open(inputReader);
List<CustomerCredit> inputs = getCredits(inputReader);
close(inputReader);
JobExecution jobExecution = this.launchJob();
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
@SuppressWarnings("unchecked")
ItemReader<CustomerCredit> outputReader = (ItemReader<CustomerCredit>) getApplicationContext().getBean(
"outputTestReader");
open(outputReader);
List<CustomerCredit> outputs = getCredits(outputReader);
close(outputReader);
assertEquals(inputs.size(), outputs.size());
int itemCount = inputs.size();
assertTrue(itemCount > 0);
for (int i = 0; i < itemCount; i++) {
assertEquals(inputs.get(i).getCredit().add(CustomerCreditIncreaseProcessor.FIXED_AMOUNT).intValue(),
outputs.get(i).getCredit().intValue());
}
}
/**
* Read all credits using the provided reader.
*/
private List<CustomerCredit> getCredits(ItemReader<CustomerCredit> reader) throws Exception {
CustomerCredit credit;
List<CustomerCredit> result = new ArrayList<CustomerCredit>();
while ((credit = reader.read()) != null) {
result.add(credit);
}
return result;
}
/**
* Open the reader if applicable.
*/
private void open(ItemReader<?> reader) {
if (reader instanceof ItemStream) {
((ItemStream) reader).open(new ExecutionContext());
}
}
/**
* Close the reader if applicable.
*/
private void close(ItemReader<?> reader) {
if (reader instanceof ItemStream) {
((ItemStream) reader).close();
}
}
}

10
spring-batch-samples/src/test/resources/org/springframework/batch/sample/PartitionJobFunctionalTests-context.xml Normal file
View File

@@ -0,0 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<import resource="classpath:/simple-job-launcher-context.xml" />
<import resource="classpath:/jobs/partitionJob.xml" />
</beans>

21
spring-batch-test/src/main/java/org/springframework/batch/test/AbstractJobTests.java
View File

@@ -32,8 +32,10 @@ import org.springframework.batch.core.job.SimpleJob;
import org.springframework.batch.core.job.flow.FlowJob;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
/**
* <p>
@@ -62,7 +64,7 @@ import org.springframework.context.ApplicationContext;
* @author Dan Garrette
* @since 2.0
*/
public abstract class AbstractJobTests {
public abstract class AbstractJobTests implements ApplicationContextAware {
/** Logger */
protected final Log logger = LogFactory.getLog(getClass());
@@ -78,6 +80,23 @@ public abstract class AbstractJobTests {
private StepRunner stepRunner;
private ApplicationContext applicationContext;
/**
* {@inheritDoc}
*/
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
/**
* @return the applicationContext
*/
protected ApplicationContext getApplicationContext() {
return applicationContext;
}
/**
* @return the job repository which is autowired by type
*/