spring-batch ItemReader用于Elasticsearch扫描和滚动

Question

在spring-batch中，是否支持使用扫描和滚动功能的Elasticsearch ItemReader？我确实看到了this扩展，但这是基于普通的spring数据搜索查询。有一个基于扫描和滚动功能的将是很好的，因为批处理作业主要需要处理大量的数据。谢谢。

Answer 1

虽然没有针对ElasticSearch的“原生”ItemReader实现，但Spring Batch确实提供了包装Spring Data PagingAndSortingRepository的RepositoryItemReader。这样，您就可以使用Spring Data ElasticSearch项目提供的ElasticSearch存储库定义。

您可以在Spring批处理文档中阅读更多关于RepositoryItemReader的内容：http://docs.spring.io/spring-batch/trunk/apidocs/org/springframework/batch/item/data/RepositoryItemReader.html。

您可以在这里阅读更多关于Spring Data ElasticSearch项目的信息：http://docs.spring.io/spring-data/elasticsearch/docs/current/reference/html/

Answer 2

import java.util.Iterator;

import org.springframework.batch.item.data.AbstractPaginatedDataItemReader;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.query.SearchQuery;

public class ElasticsearchItemReader<T> extends AbstractPaginatedDataItemReader<T> {
    private ElasticsearchOperations elasticsearchOperations;
    private final SearchQuery searchQuery;
    private String scrollId;
    private int scrollTimeinMillis = 60000;
    private Class<T> type;

    public ElasticsearchItemReader(
        final ElasticsearchOperations elasticsearchOperations,
        final SearchQuery searchQuery,
        final Class<T> type
    ) {
        this.elasticsearchOperations = elasticsearchOperations;
        this.searchQuery = searchQuery;
        this.type = type;
    }

    @Override
    protected void doOpen() throws Exception {
        scrollId = elasticsearchOperations.scan(searchQuery, scrollTimeinMillis, false);
    }

    @Override
    protected Iterator<T> doPageRead() {
        return elasticsearchOperations.scroll(scrollId, scrollTimeinMillis, type).iterator();
   }
}