Partitioner Refactor (#2434)

* Refactor Batch * Refactor Batch
2017-08-14 11:00:51 +02:00
parent 290e759d4a
commit 43357b0809
10 changed files with 49 additions and 52 deletions
@@ -0,0 +1,32 @@
+package org.baeldung.batch;
+
+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 App {
+    public static void main(final String[] args) {
+        // Spring Java config
+        final AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+        context.register(SpringConfig.class);
+        context.register(SpringBatchConfig.class);
+        context.refresh();
+
+        // Spring xml config
+        // ApplicationContext context = new ClassPathXmlApplicationContext("spring-batch.xml");
+
+        final JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
+        final Job job = (Job) context.getBean("firstBatchJob");
+        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");
+        }
+    }
+}
@@ -0,0 +1,88 @@
+package org.baeldung.batch;
+
+import org.baeldung.batch.model.Transaction;
+import org.baeldung.batch.service.CustomItemProcessor;
+import org.baeldung.batch.service.RecordFieldSetMapper;
+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.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
+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.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.io.Resource;
+import org.springframework.oxm.Marshaller;
+import org.springframework.oxm.jaxb.Jaxb2Marshaller;
+
+import java.net.MalformedURLException;
+import java.text.ParseException;
+
+public class SpringBatchConfig {
+    @Autowired
+    private JobBuilderFactory jobs;
+
+    @Autowired
+    private StepBuilderFactory steps;
+
+    @Value("input/record.csv")
+    private Resource inputCsv;
+
+    @Value("file:xml/output.xml")
+    private Resource outputXml;
+
+    @Bean
+    public ItemReader<Transaction> itemReader() throws UnexpectedInputException, ParseException {
+        FlatFileItemReader<Transaction> reader = new FlatFileItemReader<Transaction>();
+        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
+        String[] tokens = {"username", "userid", "transactiondate", "amount"};
+        tokenizer.setNames(tokens);
+        reader.setResource(inputCsv);
+        DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<Transaction>();
+        lineMapper.setLineTokenizer(tokenizer);
+        lineMapper.setFieldSetMapper(new RecordFieldSetMapper());
+        reader.setLinesToSkip(1);
+        reader.setLineMapper(lineMapper);
+        return reader;
+    }
+
+    @Bean
+    public ItemProcessor<Transaction, Transaction> itemProcessor() {
+        return new CustomItemProcessor();
+    }
+
+    @Bean
+    public ItemWriter<Transaction> itemWriter(Marshaller marshaller) throws MalformedURLException {
+        StaxEventItemWriter<Transaction> itemWriter = new StaxEventItemWriter<Transaction>();
+        itemWriter.setMarshaller(marshaller);
+        itemWriter.setRootTagName("transactionRecord");
+        itemWriter.setResource(outputXml);
+        return itemWriter;
+    }
+
+    @Bean
+    public Marshaller marshaller() {
+        Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
+        marshaller.setClassesToBeBound(Transaction.class);
+        return marshaller;
+    }
+
+    @Bean
+    protected Step step1(ItemReader<Transaction> reader, ItemProcessor<Transaction, Transaction> processor, ItemWriter<Transaction> writer) {
+        return steps.get("step1").<Transaction, Transaction>chunk(10).reader(reader).processor(processor).writer(writer).build();
+    }
+
+    @Bean(name = "firstBatchJob")
+    public Job job(@Qualifier("step1") Step step1) {
+        return jobs.get("firstBatchJob").start(step1).build();
+    }
+
+}
@@ -0,0 +1,78 @@
+package org.baeldung.batch;
+
+import java.net.MalformedURLException;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+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.support.transaction.ResourcelessTransactionManager;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+import org.springframework.jdbc.datasource.init.DataSourceInitializer;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.transaction.PlatformTransactionManager;
+
+@Configuration
+@EnableBatchProcessing
+public class SpringConfig {
+
+    @Value("org/springframework/batch/core/schema-drop-sqlite.sql")
+    private Resource dropReopsitoryTables;
+
+    @Value("org/springframework/batch/core/schema-sqlite.sql")
+    private Resource dataReopsitorySchema;
+
+    @Bean
+    public DataSource dataSource() {
+        DriverManagerDataSource dataSource = new DriverManagerDataSource();
+        dataSource.setDriverClassName("org.sqlite.JDBC");
+        dataSource.setUrl("jdbc:sqlite:repository.sqlite");
+        return dataSource;
+    }
+
+    @Bean
+    public DataSourceInitializer dataSourceInitializer(DataSource dataSource) throws MalformedURLException {
+        ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator();
+
+        databasePopulator.addScript(dropReopsitoryTables);
+        databasePopulator.addScript(dataReopsitorySchema);
+        databasePopulator.setIgnoreFailedDrops(true);
+
+        DataSourceInitializer initializer = new DataSourceInitializer();
+        initializer.setDataSource(dataSource);
+        initializer.setDatabasePopulator(databasePopulator);
+
+        return initializer;
+    }
+
+    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 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;
+    }
+
+}
@@ -0,0 +1,54 @@
+package org.baeldung.batch.model;
+
+import java.util.Date;
+
+import javax.xml.bind.annotation.XmlRootElement;
+
+@SuppressWarnings("restriction")
+@XmlRootElement(name = "transactionRecord")
+public class Transaction {
+    private String username;
+    private int userId;
+    private Date transactionDate;
+    private double amount;
+
+    /* getters and setters for the attributes */
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public int getUserId() {
+        return userId;
+    }
+
+    public void setUserId(int userId) {
+        this.userId = userId;
+    }
+
+    public Date getTransactionDate() {
+        return transactionDate;
+    }
+
+    public void setTransactionDate(Date transactionDate) {
+        this.transactionDate = transactionDate;
+    }
+
+    public double getAmount() {
+        return amount;
+    }
+
+    public void setAmount(double amount) {
+        this.amount = amount;
+    }
+
+    @Override
+    public String toString() {
+        return "Transaction [username=" + username + ", userId=" + userId + ", transactionDate=" + transactionDate + ", amount=" + amount + "]";
+    }
+
+}
@@ -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.batch.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;
+    }
+
+}
@@ -0,0 +1,166 @@
+package org.baeldung.batch.partitioner;
+
+import org.baeldung.batch.model.Transaction;
+import org.baeldung.batch.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.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;
+
+import javax.sql.DataSource;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.text.ParseException;
+
+@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<>();
+        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<>();
+        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(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 factory.getObject();
+    }
+
+    private DataSource dataSource() {
+        EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
+        return builder.setType(EmbeddedDatabaseType.HSQL)
+          .addScript("classpath:org/springframework/batch/core/schema-drop-h2.sql")
+          .addScript("classpath:org/springframework/batch/core/schema-h2.sql")
+          .build();
+    }
+
+    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;
+    }
+}
@@ -0,0 +1,28 @@
+package org.baeldung.batch.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");
+        }
+    }
+}
@@ -0,0 +1,12 @@
+package org.baeldung.batch.service;
+
+import org.baeldung.batch.model.Transaction;
+import org.springframework.batch.item.ItemProcessor;
+
+public class CustomItemProcessor implements ItemProcessor<Transaction, Transaction> {
+
+    public Transaction process(Transaction item) {
+        System.out.println("Processing..." + item);
+        return item;
+    }
+}
@@ -0,0 +1,35 @@
+package org.baeldung.batch.service;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+
+import org.baeldung.batch.model.Transaction;
+import org.springframework.batch.item.file.mapping.FieldSetMapper;
+import org.springframework.batch.item.file.transform.FieldSet;
+import org.springframework.validation.BindException;
+
+public class RecordFieldSetMapper implements FieldSetMapper<Transaction> {
+
+    public Transaction mapFieldSet(FieldSet fieldSet) throws BindException {
+
+        SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy");
+        Transaction transaction = new Transaction();
+        // you can either use the indices or custom names
+        // I personally prefer the custom names easy for debugging and
+        // validating the pipelines
+        transaction.setUsername(fieldSet.readString("username"));
+        transaction.setUserId(fieldSet.readInt("userid"));
+        transaction.setAmount(fieldSet.readDouble(3));
+        // Converting the date
+        String dateString = fieldSet.readString(2);
+        try {
+            transaction.setTransactionDate(dateFormat.parse(dateString));
+        } catch (ParseException e) {
+            e.printStackTrace();
+        }
+
+        return transaction;
+
+    }
+
+}