Step Partitioning

Time needed to complete  ~75 minutes

Prerequisite for this tutorial is Configure Spring Batch Admin to use MySQL

1.Introduction

In the situation where single threaded batch job can’t finish the work in a given time interval and tuning  JVM didn't do the trick, then it is time for scaling the batch job using multithreading. In spring batch partitioning is one way for scaling batch jobs that can improve performance.  

2.What is used in this tutorial:

1- Maven 3

2- Jdk 1.7

3- Tomcat 7.0.55

4- Eclipse Luna 4.4

5- Spring Core 3.2.9

6- Spring Batch 2.2.7

7- Spring Batch Admin Manager 1.3.0

8- MySQL 5 Database

3.Project Structure

Picture 3.1: Project structure

4. Configuration and Other files

- The database configuration where the data will be written is in file database.xml path src/main/resources/META-INF/spring/batch/dbConfig/database.xml, this is not the same as the database used by spring batch admin to store job repository data. First create the database named company_db and in the database.xml change db_username and db_password to suite company_db username and password.

Listing 4.1: database.xml

<beans xmlns="http://www.springframework.org/schema/beans"  
      xmlns:jdbc="http://www.springframework.org/schema/jdbc"   
      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-3.2.xsd  
           http://www.springframework.org/schema/jdbc   
           http://www.springframework.org/schema/jdbc/spring-jdbc-3.2.xsd">  
   <!-- connect to database -->  
      <bean id="dataSource2"  
           class="org.springframework.jdbc.datasource.DriverManagerDataSource">  
           <property name="driverClassName" value="com.mysql.jdbc.Driver" />  
           <property name="url" value="jdbc:mysql://localhost:3306/company_db" />  
           <property name="username" value="db_username" /> 
           <property name="password" value="db_password" />
      </bean>  
      <bean id="transactionManager"  
           class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />  
 </beans>  

- The job definition that will read from csv files,process and write in database table is in src/main/resources/META-INF/spring/batch/jobs/job-order-multithreading.xml , Listing 4.2. The job has two steps, organizeFilesStep step will create tmp folder in mt_data folder that will contains folders with names as number of partitions then in each folder a number of csv files will be copyed(see DistributeFiles.class). Picture 4.1.

Listing 4.2: Job definition in job-order-multithreading

 <beans xmlns="http://www.springframework.org/schema/beans"  
      xmlns:batch="http://www.springframework.org/schema/batch" xmlns:task="http://www.springframework.org/schema/task"  
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
      xsi:schemaLocation="http://www.springframework.org/schema/batch  
           http://www.springframework.org/schema/batch/spring-batch-2.2.xsd  
           http://www.springframework.org/schema/beans   
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd">  
   
      <!-- Database configuration -->  
      <import resource="../dbConfig/database.xml" />  
   
      <bean id="orderSkipListener" class="com.web.listener.OrderRecordSkipHandler" />  
      <bean id="validator" class="org.springframework.validation.beanvalidation.LocalValidatorFactoryBean" />  
      <bean id="partitioner" class="com.web.partition.MakePartition" />  
      <bean id="asyncTaskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />  
      <bean id="distributeFiles" class="com.web.general.DistributeFiles" scope="step">  
           <property name="dataDirectory" value="classpath:cvs/mt_data" />  
           <property name="numberOfPartitions" value="#{jobParameters['numberOfPartitions']}" />  
      </bean>  
   
      <!-- Batch job -->  
      <batch:job id="orderMTJob">  
   
           <batch:step id="organizeFilesStep" next="step1.boss">  
                <batch:tasklet ref="distributeFiles" />                 
           </batch:step>  
   
           <batch:step id="step1.boss">  
                <batch:partition step="worker" partitioner="partitioner" handler="orderPartitionHandler" />  
                     <!-- batch:handler grid-size="3" task-executor="asyncTaskExecutor"/>  
                </batch:partition -->  
           </batch:step>  
   
      </batch:job>  
   
      <batch:step id="worker">  
           <batch:tasklet>  
                <batch:chunk reader="filesReader" processor="orderProcessor"  
                     writer="mysqlItemWriter" commit-interval="15" skip-limit="20">  
                     <batch:skippable-exception-classes>  
                          <batch:include class="java.lang.Exception" />  
                     </batch:skippable-exception-classes>  
                </batch:chunk>  
                <batch:listeners>  
                     <batch:listener ref="orderSkipListener" />  
                </batch:listeners>  
           </batch:tasklet>  
      </batch:step>  
        
      <bean id="orderPartitionHandler" class="org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler" scope="step">  
   <property name="taskExecutor" ref="asyncTaskExecutor"/>  
   <property name="step" ref="worker" />  
   <property name="gridSize" value="#{jobParameters['numberOfPartitions']}" />  
 </bean>  
   
   
      <bean id="orderProcessor" class="com.web.processor.OrderProcessorMT" scope="step">  
           <property name="validator" ref="validator" />            
      </bean>  
        
       <bean id="filesReader"     class=" org.springframework.batch.item.file.MultiResourceItemReader" scope="step">  
      <property name="resources" value="classpath:cvs/mt_data/tmp/#{stepExecutionContext['partitionNo']}/#{jobParameters['inputDataFileFormat']}" />  
      <property name="delegate" ref="cvsFileItemReader" />  
  </bean>  
    
      <bean id="cvsFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="prototype">  
           <property name="linesToSkip" value="1" />  
           <!-- Read order csv file -->  
             
           <property name="lineMapper">  
                <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">  
   
                     <!-- Spliting fileds -->  
                     <property name="lineTokenizer">  
                          <bean  
                               class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer" />  
                     </property>  
                     <property name="fieldSetMapper">  
                          <bean class="com.web.mapper.OrderMapper" />  
                     </property>  
                </bean>  
           </property>  
      </bean>       
   
      <bean id="mysqlItemWriter"  
           class="org.springframework.batch.item.database.JdbcBatchItemWriter" scope="prototype">  
           <property name="dataSource" ref="dataSource2" />  
           <property name="sql">  
                <value>  
       <![CDATA[      
            insert into tel_order(code,service_type,quantity,offer,bill_type,order_date)  
         values (:code, :serviceType, :quantity, :offer, :billType,:orderDate)  
       ]]>  
                </value>  
           </property>  
           <!-- map java object property to sql parameter -->  
           <property name="itemSqlParameterSourceProvider">  
                <bean  
                     class="org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider" />  
           </property>  
      </bean>  
   
 </beans>  
   

Picture 4.1: organizeFilesStep will copy files in emp/number_of_partition folder

- The second step is step1.boss , picture 4.2, in this step 3 threads (grid-size) will be created, where thread1 will read csv files only from tmp/1 folder, thread2 will read only from tmp/2 folder and thread3 will read only from tmp/3 folder. Each thread will have it's own reader,processor and writer. Readers and Writers bean have scope attribute prototype , and it is a good place to learn what happened if scope="prototype" is removed. Try it and see the data race to lineNumber in reader. Reader by default are singleton's without prototype scope there will be one reader instance and three threads that reading and updating the next lineNumber to be read, causing incorrect behavior. An important parameter is commit-interval changing it's value will have a high impact on the performance, e.g.  value 3 will degenerate performance. 

Picture 4.2: orderMTJob Steps

- The Order class listing 4.3.

Listing 4.3: Order class

 package com.web.model;  
   
 import java.io.Serializable;  
 import java.util.Date;  
   
 import javax.validation.constraints.Max;  
 import javax.validation.constraints.Min;  
 import javax.validation.constraints.NotNull;  
 import javax.validation.constraints.Past;  
 import javax.validation.constraints.Size;  
   
 import org.springframework.format.annotation.DateTimeFormat;  
   
 import com.web.validator.AcceptedValues;  
   
   
 public class Order implements Serializable {  
   
      private static final long serialVersionUID = 1853355651832301663L;  
   
      @Size(min = 5, max = 5, message = "Invalid code value")  
      private String code;  
        
      @AcceptedValues(acceptValues = { "fix", "mobile", "bundle", "internet" }, message = "Invalid service type")  
      private String serviceType;  
   
      @Min(value = 1, message = "Quantity less than 1")  
      @Max(value = 9, message = "Quantity more than 9")  
      private int quantity;  
        
      @AcceptedValues(acceptValues = { "small", "medium", "large", "fiber32" }, message = "Invalid offer")  
      private String offer;  
   
      @AcceptedValues(acceptValues = { "prepaid", "postpaid_mail","postpaid_email" }, message = "Invalid bill type")  
      private String billType;  
   
      @NotNull(message = "Date is required")  
      @Past(message = "Order date should be in the past")  
      @DateTimeFormat(pattern = "dd-MM-yyyy")  
      private Date orderDate; 

    // Setter, getters, toString, etc omitted    

Listing 4.4: Order mapper

 package com.web.mapper;  
   
 import org.springframework.batch.item.file.mapping.FieldSetMapper;  
 import org.springframework.batch.item.file.transform.FieldSet;  
 import org.springframework.validation.BindException;  
   
 import com.web.model.Order;  
   
 public class OrderMapper implements FieldSetMapper<Order>{  
   
      @Override  
      public Order mapFieldSet(FieldSet fieldSet) throws BindException {  
           Order order = new Order();  
             
           order.setCode(fieldSet.readString(0));  
           order.setServiceType(fieldSet.readString(1));  
           order.setQuantity(fieldSet.readInt(2));  
           order.setOffer(fieldSet.readString(3));  
           order.setBillType(fieldSet.readString(4));  
           order.setOrderDate(fieldSet.readDate(5, "dd-MM-yyyy"));  
             
           return order;  
      }  
   
 }  
   

- MakePartition, Listing 4.5.

Listing 4.5: orderMTJob Steps

 package com.web.partition;  
   
 import java.util.HashMap;  
 import java.util.Map;  
   
 import org.springframework.batch.core.partition.support.Partitioner;  
 import org.springframework.batch.item.ExecutionContext;  
   
 public class MakePartition implements Partitioner {  
   
      @Override  
      public Map<String, ExecutionContext> partition(int gridSize) {  
             
           Map<String, ExecutionContext> executionContextMap = new HashMap<String, ExecutionContext>();  
             
           for (int i = 1; i <= gridSize; i++) {  
                  
                ExecutionContext executionContext = new ExecutionContext();                 
                executionContext.putString("threadId", "Thread_"+i);  
                executionContext.putInt("partitionNo", i);                 
                executionContextMap.put("Partition_"+i, executionContext);  
           }  
             
           return executionContextMap;  
      }  
   
 }  
   

- SQL for creating table company_db.tel_order

Listing 4.6: SQL for creating tel_order table

 CREATE TABLE tel_order (  
  id INTEGER(11) NOT NULL AUTO_INCREMENT,  
  code VARCHAR(5) NOT NULL,  
  service_type varchar(32) NOT NULL,  
  quantity INTEGER(11) NOT NULL,  
  offer VARCHAR(32) NOT NULL,  
  bill_type VARCHAR(32) NOT NULL,  
  order_date timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  PRIMARY KEY (id)  
 ) ENGINE=InnoDB;  

5. CSV files

- The are 100 csv input files in folder mt_data containing 1900 orders.

Listing 5.1:order data

 code,service_type,quantity,offer,bill_type,order_date  
 234ac,mobile,9,small,prepaid,12-08-2014  
 235ab,fix,1,medium,postpaid_mail,14-08-2014  
 237ar,bundle,1,small,postpaid_email,10-08-2014  
 236as,bundle,1,small,postpaid_email,12-08-2014  
 239ac,fix,1,large,postpaid_mail,12-08-2014  
 230mb,mobile,2,small,prepaid,12-08-2014  
 244rx,mobile,1,large,postpaid_email,12-08-2014  
 274cb,mobile,5,medium,prepaid,12-08-2014  
 228wb,mobile,1,large,postpaid_email,12-07-2014  
 284mn,mobile,1,small,prepaid,12-08-2014  
 285rn,fix,1,large,prepaid,12-08-2014  
 292wr,bundle,1,medium,postpaid_email,12-08-2014  
 293un,mobile,1,medium,prepaid,12-08-2014  
 304po,bundle,1,large,postpaid_email,12-08-2014  
 305is,bundle,1,large,postpaid_email,12-08-2014  
 311jr,bundle,1,small,postpaid_mail,12-08-2014  
 312ps,internet,1,fiber32,postpaid_email,12-08-2014  
 317tr,mobile,3,small,prepaid,12-08-2014  
 329qv,mobile,1,medium,prepaid,12-08-2014  

6. Running orderMTJob in eclipse

- To run the application use code in class src/main/java/com/web/app/MultithreadingOrderApp.java Listing 6.1. Changing  jobParameters  numberOfPartitions(grid-size) will change the execution time for the batch job, bigger i not better, it depends on the number of threads(logical processor) in the processor. Example 8 cores-16 threads processor.

Listing 6.1: MultithreadingOrderAppClass

 package com.web.app;  
   
 import java.util.Date;  
   
 import org.apache.log4j.Logger;  
 import org.springframework.batch.core.Job;  
 import org.springframework.batch.core.JobExecution;  
 import org.springframework.batch.core.JobParameters;  
 import org.springframework.batch.core.JobParametersBuilder;  
 import org.springframework.batch.core.launch.JobLauncher;  
 import org.springframework.context.support.ClassPathXmlApplicationContext;  
   
 public class MultithreadingOrderApp {  
   
      private static final Logger log = Logger.getLogger(MultithreadingOrderApp.class);  
      private static ClassPathXmlApplicationContext context;  
   
      public static void main(String[] args) {  
   
           String exitStatus = "Exit Status: ";  
   
           String[] springConfig = { "launch-context.xml",  
                     "META-INF/spring/batch/jobs/job-order-multithreading.xml" };  
   
           context = new ClassPathXmlApplicationContext(springConfig);  
   
           JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");  
           Job job = (Job) context.getBean("orderMTJob");  
   
           try {  
                JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();  
   
                jobParametersBuilder.addString("dateTime", (new Date()).toString());  
                jobParametersBuilder.addString("numberOfPartitions", "3");       
                jobParametersBuilder.addString("inputDataFileFormat", "orders_*.csv");  
                  
   
                JobParameters param = jobParametersBuilder.toJobParameters();  
                  
                long t1 = System.currentTimeMillis();  
                JobExecution execution = jobLauncher.run(job, param);  
                log.info("TIME:"+(System.currentTimeMillis()- t1));  
                  
   
                exitStatus += execution.getStatus().toString();  
   
           } catch (Exception e) {  
                log.info(e.getStackTrace());  
           } finally {  
                if (context != null)  
                     context.close();  
           }  
   
           log.info(exitStatus);  
           log.info("Job Finished");  
   
      }  
   
 }  
  

7. Running orderJob From Spring Batch Admin

- In pom.xml change username_tc and password_tc to suite your local tomcat. 

- Start tomcat.

- Deploy application from run configuration in goal type clean install tomcat7:redeploy , click apply and 

Run buttons.

Note: If these steps are unfamiliar read steps 4 and 5 deploying spring batch admin Link How To

- open address localhost:8080 /springBatchAdminMysql/jobs  in web browser , picture 7.1

Picture 7.1: Jobs list

- Click orderMTJob, Change dateTime parameter, click launch button.  job will completed with COMPLETED status. picture 7.2. orderMTJob steps.

Picture 7.2:orderMTJob steps status

- Click on executions -> ID number of job execution -> COMPLETED under column Status picture 7.3.

Picture 7.3: step1.boss Step Summary View

8. Eclipse Project

 Download project source code