This is continuation to post. In General, the batch processing needs to process tons tons of data instead of running simple tasks(as we saw in last post). Spring batch allows us to do this with Reader/Writer and Processors. As the name suggests, reader will read the data from the source, processor processes the source data and converts to output. Finally, Writers writes to the destination.
Looks simple, then how can we achieve chunk oriented processing with this. Spring simplified the process to the developers. We need just take care of from where to read(Reader), how to process it(Processor) and where and how to write the processed data(Writer).
Now will see the following with example:
Setup a job with steps ( This we already know)
Configure readers, writers and processor for each step
firstStep - A combination of testReader, testWriter and testProcessor with commit-interval of 5. Once the step is completed, the job proceeds to secondStep (because next attribute is defined as secondStep). The commit-interval says that - five items will be read one by one and processed but the 5 items will be written at once
secondStep - Same as above but commit interval is 2
anotherJob - Consists of three steps
failStep - A combination of failedTestReader, testWriter and testProcessor with a commit-interval of 2.
Goes to sucessStep when completed properly
Goes to failedStep when there is an exception.
sucessStep - Job executes this step for all cases except when failed
failedStep - Job executes this step if the first step is failed or throws an exception.
Step can be defined as Tasklet(Simple task) or a combination of Reader, Writer and Processor (Chunk oriented processing).
We can define next step either
As attribute of step tag - The next step will be executed irrespective of the success criteria
Separate tag inside step - The next step will be executed based on the outcome of current step.
Reader, Writer and Processor should implement ItemReader, ItemWriter and ItemProcessor respectively. (As shown in the next code snippet).
ItemReader has method read - which returns a generic type I (Should be an item which is to be read)
ItemProcessor has method process - takes I as an input (which is read by Reader) and returns O (Should be an item which should be committed).
ItemWriter has method write - takes O as an input which is processed by Processor.
ItemWriter has the write method which takes list as an argument - Because commit will be done in batch (batch count will be equal to commit-interval).
Reader reads continuously (in a step), unless it returns a null. One reader returns null, the step will be completed after committing the batch.
Writer commits the data once the reader read n (commit-interval) number of inputs and processor processed them or reader completed reading all the inputs (returns null).
When an exception occurs, writer ignores the data to write. This case need to handled explicitly - (Using failed step in the above XML).
The below is the code representing the above jobs and their reader, writer, processor and tasklets with input and output data objects.
Source Source.java - Which is used an input - prepared by Reader
package com.test.entity;
public class Source {
private String input;
//getters and setters
}
Destination.java - Which is prepared by Processor and used by Writer
package com.test.entity;
public class Destination {
private String output;
//getters and setters
}
TestReader.java
package com.test.batch;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import com.test.entity.Source;
public class TestReader implements ItemReader<Source> {
private int count = 10;
private boolean raiseError = false;
private static int internalCount = 0;
public Source read() throws Exception, UnexpectedInputException,
ParseException, NonTransientResourceException {
if(raiseError)
{
System.out.println("Exception occured while reading");
throw new Exception("New Exception");
}
internalCount++;
if(internalCount >= count)
{
System.out.println("Reached max count "+internalCount);
return null;
}
Source source = new Source();
source.setInput("Input "+internalCount);
System.out.println("Reading item "+internalCount);
return source;
}
//Getters and Setters
}
TestProcessor.java
package com.test.batch;
import org.springframework.batch.item.ItemProcessor;
import com.test.entity.Destination;
import com.test.entity.Source;
public class TestProcessor implements ItemProcessor {
public Destination process(Source source) throws Exception {
Destination destination = new Destination();
destination.setOutput(source.getInput().replace("Input", "Output"));
System.out.println("Converted "+source.getInput()+" to "+destination.getOutput());
return destination;
}
}
TestWriter.java
package com.test.batch;
import java.util.List;
import org.springframework.batch.item.ItemWriter;
import com.test.entity.Destination;
public class TestWriter implements ItemWriter {
public void write(List arg0) throws Exception {
for(Destination dest : arg0)
{
System.out.println("Writing : "+dest.getOutput());
}
System.out.println("Finished Writing");
}
}
TestTasklet.java
package com.test.batch;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
public class TestTasklet implements Tasklet {
private boolean fail = false;
public RepeatStatus execute(StepContribution arg0, ChunkContext arg1)
throws Exception {
if (!fail) {
System.out.println("Finished sucessfully");
return RepeatStatus.FINISHED;
} else {
System.out.println("Exception... So rollback should take place");
return RepeatStatus.FINISHED;
}
}
//Getters and Setters
}
Output
When we run the job sucessJob
INFO: Executing step: [firstStep]
Reading item 1
Reading item 2
Reading item 3
Reading item 4
Reading item 5
Converted Input 1 to Output 1
Converted Input 2 to Output 2
Converted Input 3 to Output 3
Converted Input 4 to Output 4
Converted Input 5 to Output 5
Writing : Output 1
Writing : Output 2
Writing : Output 3
Writing : Output 4
Writing : Output 5
Finished Writing
Reading item 6
Reading item 7
Reading item 8
Reading item 9
Reached max count 10
Converted Input 6 to Output 6
Converted Input 7 to Output 7
Converted Input 8 to Output 8
Converted Input 9 to Output 9
Writing : Output 6
Writing : Output 7
Writing : Output 8
Writing : Output 9
Finished Writing
INFO: Executing step: [secondStep]
Reached max count 11
INFO: Job: [FlowJob: [name=sucessJob]] completed with the following parameters: [{}] and the following status: [COMPLETED]
When we run the anotherJob. Output looks like
INFO: Executing step: [failStep]
SEVERE: Encountered an error executing the step
java.lang.Exception: New Exception
at com.test.batch.TestReader.read(TestReader.java:24)
at com.test.batch.TestReader.read(TestReader.java:1)
at org.springframework.batch.core.step.item.SimpleChunkProvider.doRead(SimpleChunkProvider.java:90)
...
Exception occured while reading
Jan 02, 2014 11:34:30 PM org.springframework.batch.core.job.SimpleStepHandler handleStep
INFO: Executing step: [failedStep]
Exception... So rollback should take place
INFO: Job: [FlowJob: [name=anotherJob]] completed with the following parameters: [{}] and the following status: [COMPLETED]
Spring JMX allows you to easily expose POJOs as JMX MBeans. Even it allows the MBeans to deploy on application server which has MBean server running or can be run standalone. We can use jconsole to run the services provided by the MBean.
The configuration required to expose the POJO as Mbean is as follows. Here, MyBean is a POJO with at-least one public method.
MBeanExporter exposes the map of "beans" as MBeans. Here we exposed myMBean as "MyMMBeanName". This name will be shown in the jconsole. By default, all the public methods inside the POJO will be exposed as operations.
If the Spring JMX is not running under an application server, then we need a starter. Definition is as follows:
<!-- If not running on a server which has MBean server running, you must start here -->
<bean id="factory" class="org.springframework.jmx.support.MBeanServerFactoryBean" />
With the above configuration, The MBean can be accessed locally. The MBean can be exposed either using JMXJMP or RMI.
To expose using JMXJMP as follows. The default service URL to access is : service:jmx:jmxmp://localhost:9875
Spring batch 2.0 simpilified and improvised the batching framework. We have so many frameworks for MVC, Web, JDBC etc but batching frameworks are very rare. Spring batch is a lightweight and robust batch framework to process these big data sets. Spring offer Tasklet oriented and Chunk Oriented processing. In this post, we will see Simple Tasklet oriented processing.
Key Concepts:
Job - Job is a sequence of steps, each has an exit status. Execution of the next step depends on the exit status of previous step.
JobRepository - An interface which contains the meta data and corresponding entities of the Job.
JobLauncer - Which launches a job by exposing the method to run.
Tasklet - An interface, can be instance of job which exposes a method called execute and returns the execution status. A tasklet will execute repeatedly until it returns FINSIHED.
Step to define (A simple job):
Define one job repository
Define one job launcher using the job repository.
Define Job(s) under the job launcher. Multiple jobs can be defined under one job launcher.
Create steps under the job. We can add multiple steps under the job with relations. By default, job launcher executes the job based on the steps defined. We can create dependency between each step. Literally, each step is a java class (Either Tasklet - a simple task or a combination of reader, writer and executiors for chunk oriented processing).
The below example shows how to define a job with single task (or step) using Tasklet.
The SampleTasklet class should be a type of Tasklet, which provides one execute. The execute method returns RepeatStatus to know the status of the task. The tasks executes multiple times if it returns CONTINUABLE otherwise stops after execution.
package com.test.springbatch;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
public class SampleTasklet implements Tasklet
{
public RepeatStatus execute(StepContribution arg0, ChunkContext arg1) throws Exception
{
System.out.println("Do Something; ");
// Return RepeatStatus.CONTINUABLE if something goes wrong so that it
// repeats; Otherwise Return FINISHED to complete
return RepeatStatus.FINISHED;
}
}
How to run the job:
Spring provides you with CommandLineJobRunner which runs the job using the two parameters. One the spring context file and the other is the batch name.
A lot of effort is required for configuration and using JPA. Spring Data JPA has reduced that to a major extent. Takes little effort to configure and easy to use. Let's look at this with an example.
The spring data dependency and hibernate dependencies using maven are as follows. (Note: All other spring core, bean, context dependencies are required along with the below dependencies)
Explanation of above spring configuration
Create one datasource.
Create one driver manager datasource named "mysqltestDataSource", by loading jdbc.properties file from the classpath using line <context:property-placeholder location="classpath:jdbc.properties"/>
Create entity manager factory using the datasource created and persistent unit created in persistent.xml
Inject entity manager factory into transaction manager.
Finally configure both transaction manager and entity manager factory to set of repositories in a package (defined using base-package)
Now, create repository, an interface which extends JpaRepository inside package (defined in base-package). The typed arguments required are one entity and other is Id. In this case, entity is Employee and Id of the Employee is Long.
package com.test.entity;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;
@Table(name = "EMP")
@Entity
public class Employee
{
private int id;
private String name;
private int age;
@Id
@Column(name = "id", unique = true, nullable = false)
public int getId()
{
return id;
}
public void setId(int id)
{
this.id = id;
}
@Column(name = "name")
public String getName()
{
return name;
}
public void setName(String name)
{
this.name = name;
}
@Column(name = "age")
public int getAge()
{
return age;
}
public void setAge(int age)
{
this.age = age;
}
}
That's it on the configuration part. By using the repository, we can do all types of database operations for that entity. There are numberous in-built methods are provided by JpaRepository interface. Sample code for that will be
EmployeeDAO dao = (EmployeeDAO)context.getBean("employeeDAO");
List list = dao.findAll();
for(Employee e : list)
{
//Do Something
}
The New IO is one of the features introduced in Java7. A new package java.nio has been added as part of this feature. Even though the usage wise it's same as java.io, most of the issues in java.io has been addressed. We will see the features introduced in New IO (NIO) Path
The package java.nio.file consists of classes and interfaces Path, Paths, FileSystem, FileSystems and others. Each of represent's the file path or file system as is.
java.nio.file.Path works exactly same as java.io.File with additional features.
Files
Path can also be used for deleting. There are two delete methods. One delete method can throw NoSuchFileException if the file doesn't exist.
Files.delete(path);
Where as other deletes the file if exists.
Files.deleteIfExists(path);
There are copy, move, create (Both for directories and files) with options. There are mthods to create symbolic links, temporary directories as well.
WatchService
This is the service which we can get the file change notifications like delete, update, create, change etc. WatchService API makes to receive notifications on a file or directory.
