Trying to get to the habit of blogging often this year. New year’s resolution, hoping I can make it as a habit.
One of the interesting thing I was struggling was to try out was a dual update to two data-sources and work through a model where spring batch kicks off the entire workflow. Azure cloud has always been a good place to try out some spring related stack with its native support.
So here is the problem statement that this post is for., get a open data csv file, parse it using spring batch and update an h2 database and replicate the same data in CosmosDB.
CosmosDB is a managed serverless offering from Azure and is a very robust NoSQLDB. Here is a high level overview of the solution
The csv file for the solution can be downloaded from the open data portal. The file has lot more information than what is needed for the solution. Spring boot provides a API tool set for reading csv files. Here is the code snippet for the reader.
@Bean
public FlatFileItemReader<RoadInput> reader() {
return new FlatFileItemReaderBuilder<RoadInput>().name("MatchItemReader")
.resource(new ClassPathResource("Rustic_Roads.csv")).delimited().names(DATA_STRINGS).
fieldSetMapper(new BeanWrapperFieldSetMapper<RoadInput>() {
{
setTargetType(RoadInput.class);
}
}).build();
}
The processor for the input file that has to be converted to an entity object goes something like this
import me.sathish.myroadbatch.data.Road;
import me.sathish.myroadbatch.data.RoadInput;
import org.springframework.batch.item.ItemProcessor;
public class RoadDataProcessor implements ItemProcessor<RoadInput, Road> {
@Override
public Road process(RoadInput roadInput) throws Exception {
Road road = new Road();
road.setRoad_name(roadInput.getRoadName());
road.setSurface_type(roadInput.getSurfaceType());
road.setMiles(roadInput.getMiles());
road.setTotal_miles(roadInput.getTotalMiles());
road.setRustic_road_number(roadInput.getRusticRoadNumber());
road.setCommunity(roadInput.getCommunity());
road.setCounty(roadInput.getCounty());
road.setLine_to_web_page(roadInput.getLineToWebpage());
road.setShape_len(roadInput.getShapeLen());
return road;
}
}
The writer is a simple JDBC template write
@Bean
public JdbcBatchItemWriter<Road> writer(DataSource dataSource) {
return new JdbcBatchItemWriterBuilder<Road>()
.itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
.sql("Insert Into Road(road_name, surface_type, miles, total_miles, rustic_road_number," +
" community, county,line_to_web_page," +
" shape_len) VALUES(:road_name, :surface_type,:miles, " +
":total_miles,:rustic_road_number,:community,:county,:line_to_web_page," +
":shape_len)").dataSource(dataSource).build();
}
Finally the step and job for the reminder of the code is
@Bean
public Job importRoadUserJob(RoadsJobCompletionNotificationListener listener, Step step1) {
return jobBuilderFactory.get("importRoadUserJob").
incrementer(new RunIdIncrementer()).listener(listener).
flow(step1).end().build();
}
@Bean
public Step step1(JdbcBatchItemWriter<Road> writer) {
return stepBuilderFactory.
get("step2").<RoadInput, Road>chunk(10).
reader(reader()).
processor(processor()).
writer(writer).
build();
}
The repo write is based of spring flux., here is a repository code
package me.sathish.myroadbatch.repository;
import com.azure.spring.data.cosmos.repository.ReactiveCosmosRepository;
import me.sathish.myroadbatch.data.RoadForCosmos;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
@Repository
public interface RoadForCosmosRepo extends ReactiveCosmosRepository<RoadForCosmos, String> {
Flux<RoadForCosmos> findByMiles(String miles);
}
The properties file for the application
spring.h2.console.enabled=true
spring.datasource.url=jdbc:h2:mem:mytestdb
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=
spring.jpa.database-platform=org.hibernate.dialect.H2Dialect
spring.jpa.show-sql=true
spring.jpa.hibernate.ddl-auto=update
# Specify the DNS URI of your Azure Cosmos DB.
azure.cosmos.uri=https://cosmosdbv7.documents.azure.com:443/
# Specify the access key for your database.
azure.cosmos.key=
# Specify the name of your database.
azure.cosmos.database=roadwayscsv
azure.cosmos.populateQueryMetrics=true
finally the listener that ties the dual update is
package me.sathish.myroadbatch.listener;
import me.sathish.myroadbatch.data.Road;
import me.sathish.myroadbatch.data.RoadForCosmos;
import me.sathish.myroadbatch.repository.RoadForCosmosRepo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.persistence.EntityManager;
import java.util.List;
@Component
public class RoadsJobCompletionNotificationListener extends JobExecutionListenerSupport {
private static final Logger LOGGER = LoggerFactory.getLogger(RoadsJobCompletionNotificationListener.class);
private final EntityManager entityManager;
@Autowired
private RoadForCosmosRepo roadForCosmosRepo;
@Autowired
public RoadsJobCompletionNotificationListener(EntityManager entityManager) {
this.entityManager = entityManager;
}
@Override
public void afterJob(JobExecution jobExecution) {
LOGGER.debug("Finishing up H2 insert. Going to Couch DB");
this.roadForCosmosRepo.deleteAll().block();
if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
Object count = entityManager.createQuery("from Road").getResultList().get(0);
List<Road> roadList = entityManager.createQuery("from Road ").getResultList();
processForCosmosDB(roadList);
System.out.println("The count is " + count);
}
}
private void processForCosmosDB(List<Road> roadList) {
this.roadForCosmosRepo.deleteAll().block();
roadList.stream().forEach(road -> {
RoadForCosmos roadForCosmos = new RoadForCosmos();
roadForCosmos.setId(road.getId());
roadForCosmos.setCounty(road.getCounty());
roadForCosmos.setCommunity(road.getCommunity());
roadForCosmos.setMiles(road.getMiles());
roadForCosmos.setRoad_name(road.getRoad_name());
roadForCosmos.setLine_to_web_page(road.getLine_to_web_page());
roadForCosmos.setRustic_road_number(road.getRustic_road_number());
final RoadForCosmos data = roadForCosmosRepo.save(roadForCosmos).block();
});
}
}
The complete code base in the github url. Hope this is useful when you are looking for a solution. Remember cosmosDB is a free managed service, I had a tough time getting an instance in the US region. So I had to get a my instance in Canada region.
Sathish Jayapal 
You must be logged in to post a comment.