如何在Alpakka中实现无Scroll的Elasticsearch响应式搜索？

核心结论 #

• 无法通过设置bufferSize或scrollDuration为0禁用Scroll：Alpakka的ElasticsearchSource是专为Scroll场景设计的，源码中唯一的请求发送逻辑就是sendScrollScanRequest，设置0值会触发参数校验失败或运行时异常。

可行解决方案 #

方案1：使用ElasticsearchClient直接发送普通搜索请求 #

Alpakka提供的底层ElasticsearchClient支持发送标准的非Scroll搜索请求，你可以将返回结果转换为Akka Stream的Source来实现响应式处理。

示例代码（Scala）：

import akka.actor.ActorSystem
import akka.stream.scaladsl.Source
import org.apache.http.HttpHost
import org.elasticsearch.client.RestClient
import com.lightbend.elasticsearch.client.{ElasticsearchClient, ElasticsearchParams}

// 初始化Akka系统和Elasticsearch客户端
implicit val system: ActorSystem = ActorSystem("ElasticsearchNoScrollDemo")
val restClient = RestClient.builder(HttpHost.create("http://localhost:9200")).build()
val elasticClient = ElasticsearchClient(restClient)

// 定义搜索请求和参数
val searchQuery = """{"query": {"match_all": {}}}"""
val searchParams = ElasticsearchParams("your_index_name", "_search")

// 将异步请求结果转换为Stream Source
val searchResultsSource: Source[Map[String, Any], _] = Source
  .future(elasticClient.execute(searchQuery, searchParams))
  .flatMapConcat { response =>
    // 解析返回的hits列表
    val hits = response.bodyAs[Map[String, Any]]("hits.hits").asInstanceOf[List[Map[String, Any]]]
    Source(hits)
  }

// 处理流中的数据
searchResultsSource.runForeach(hit => println(s"搜索结果: $hit"))

方案2：自行实现基于from/size的分页流 #

如果需要处理大量数据但不想用Scroll，可以基于Elasticsearch的from和size参数实现分页查询，用Akka Stream的unfoldAsync来循环请求每页数据，直到没有更多结果。

示例代码（Scala）：

def createPaginatedSource(
  client: ElasticsearchClient,
  index: String,
  baseQuery: String,
  pageSize: Int
): Source[Map[String, Any], _] = {
  // 使用unfoldAsync迭代分页请求
  Source.unfoldAsync(0) { currentOffset =>
    // 构建带分页参数的查询
    val paginatedQuery = s"""{"query": $baseQuery, "from": $currentOffset, "size": $pageSize}"""
    val params = ElasticsearchParams(index, "_search")
    
    client.execute(paginatedQuery, params).map { response =>
      val hits = response.bodyAs[Map[String, Any]]("hits.hits").asInstanceOf[List[Map[String, Any]]]
      // 如果没有更多结果则终止迭代，否则继续下一页
      if (hits.isEmpty) None
      else Some((currentOffset + pageSize, hits))
    }
  }.flatMapConcat(Source(_)) // 将每页的hits列表展开为流元素
}

// 使用示例
val baseQuery = """{"match_all": {}}"""
val paginatedSource = createPaginatedSource(elasticClient, "your_index_name", baseQuery, 100)
paginatedSource.runForeach(hit => println(s"分页结果: $hit"))

内容的提问来源于stack exchange，提问作者sadegh