title: how partition reader from database, writer in different files and optimize thread load
body: <p>EDIT: </p>
  
  <p>I think that there are something wrong with this clause: </p>
  
  <p> </p>
  
  <p>I tried to run my first test that runs single thread and take about 35 minutes with this whereCause and the execution is terribly slow. When I just do an select * from table, whitout whereClause the process happens normally. </p>
  
  <hr>
  
  <p>I trying to use Step Partitioning in a Job with Spring Batch, but I dont realize if is it's 
  appropriate to my case: </p>
  
  <p>I have read from a database with ~30 million records. In the record, I have a column bank_id and there is about 23 differents banks. </p>
  
  <p>I have to read the value from this column and separate the records from each bank into different txt files. </p>
  
  <p>I want my job parallelize the work in 4 or 8 threads, in a first moment I try to use step partitioning and I split the job in 4 slaves and set the id_bank that I process in a parameter for a query in SqlPagingQueryProviderFactoryBean and I use only 4 different Ids. But the amount of records from one bank_id to another varies widely resulting in a slave finish they job before anothers. </p>
  
  <p>I want that when the slave finish they work, he begin to process another bank_id.</p>
  
  <p>I need a help to do anything like this in spring batch. I use the 2.1 version of spring batch. </p>
  
  <p>here is my files:</p>
  
  <pre><code>&lt;bean id="arquivoWriter"
          class="org.springframework.batch.item.file.FlatFileItemWriter"
          scope="step"&gt;
          &lt;property name="encoding" value="ISO-8859-1" /&gt;
          &lt;property name="lineAggregator"&gt;
              &lt;bean
                  class="org.springframework.batch.item.file.transform.FormatterLineAggregator"&gt;
                  &lt;property name="fieldExtractor"&gt;
                      &lt;bean
                          class="org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor"&gt;
                          &lt;property name="names"
                              value="name_bank, id_bank, etc" /&gt;
                      &lt;/bean&gt;
                  &lt;/property&gt;
                  &lt;property name="format"
                      value="..." /&gt;
              &lt;/bean&gt;
          &lt;/property&gt;
          &lt;property name="resource"
              value="file:./arquivos/#{stepExecutionContext[faixa]}.txt" /&gt;
      &lt;/bean&gt;
  
  
  
  &lt;job id="partitionJob" xmlns="http://www.springframework.org/schema/batch"&gt;
      &lt;step id="masterStep"&gt; 
          &lt;partition step="slave" partitioner="rangePartitioner"&gt;
              &lt;handler task-executor="taskExecutor" /&gt;
          &lt;/partition&gt;
      &lt;/step&gt;
  &lt;/job&gt;
  
  &lt;step id="slave" xmlns="http://www.springframework.org/schema/batch"&gt;
      &lt;tasklet&gt;
          &lt;chunk reader="pagingReader" writer="arquivoWriter"
              commit-interval="#{jobParameters['commit.interval']}" /&gt;
  
          &lt;listeners&gt;
              &lt;listener ref="myChunkListener"&gt;&lt;/listener&gt;
          &lt;/listeners&gt;        
      &lt;/tasklet&gt;
  &lt;/step&gt;
  
  &lt;bean id="rangePartitioner" class="....RangePartitioner" /&gt;
  
  
  
  &lt;bean id="pagingReader"
      class="org.springframework.batch.item.database.JdbcPagingItemReader"
      scope="step"&gt;
      &lt;property name="dataSource" ref="dataSource" /&gt;
      &lt;property name="fetchSize" value="#{jobParameters['fetch.size']}"&gt;&lt;/property&gt;
      &lt;property name="queryProvider"&gt;
          &lt;bean
              class="org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean"&gt;
              &lt;property name="dataSource" ref="dataSource" /&gt;
              &lt;property name="selectClause"&gt;
                  &lt;value&gt;
                      &lt;![CDATA[
                         SELECT ...
                      ]]&gt;
                  &lt;/value&gt;
              &lt;/property&gt;
              &lt;property name="fromClause" value="FROM my_table" /&gt;
              &lt;property name="whereClause" value="where id_bank = :id_op" /&gt; 
  
          &lt;/bean&gt;
      &lt;/property&gt;
      &lt;property name="parameterValues"&gt;
          &lt;map&gt;
              &lt;entry key="id_op" value="#{stepExecutionContext[id_op]}" /&gt;
          &lt;/map&gt;
      &lt;/property&gt;
      &lt;property name="maxItemCount" value="#{jobParameters['max.rows']}"&gt;&lt;/property&gt;
      &lt;property name="rowMapper"&gt;
          &lt;bean class="....reader.MyRowMapper" /&gt;
      &lt;/property&gt;
  &lt;/bean&gt;
  </code></pre>
  
  <p>The range partitioner: </p>
  
  <pre><code>public class RangePartitioner implements Partitioner {
  
  @Autowired
  BancoDao bancoDao;
  
  final Map&lt;String, ExecutionContext&gt; result = new HashMap&lt;String, ExecutionContext&gt; ();
  @Override
  public Map&lt;String, ExecutionContext&gt; partition(int gridSize) {
      List&lt;OrgaoPagadorQuantidadeRegistrosTO&gt; lista = bancoDao.findIdsOps();
      for (OrgaoPagadorQuantidadeRegistrosTO op:lista){
          String name = String.valueOf(op.getIdOrgaoPagador());
          ExecutionContext ex = new ExecutionContext();
          ex.putLong("id_op", op.getIdBank());
          ex.putString ("faixa", name);
          result.put("p"+name, ex);
      }
      return result;
  }
  </code></pre>
  
  <p>}</p>