c++ 实现滚动查询Elasticsearch

[hadoop@iZ25s7cmfyrZ C_script]$ cat cpp_elasticsearch.cpp
#include <iomanip>
#include <iostream>
#include <cctype>
#include <string>
#include <vector>
#include <sstream>
#include <boost/lexical_cast.hpp>
#include <json/json.h>
#include <boost/network/protocol/http/client.hpp>

namespace http=boost::network::http;
namespace network=boost::network;
using std::cout;
using std::endl;
using std::string;

typedef std::pair<string,Json::Value> SJPair;
typedef std::vector<string> StrVec;

string url_encode(const string &rhs){
std::ostringstream oss;
std::setfill('0');
oss << std::hex;
for(string::const_iterator cit=rhs.begin();
cit!=rhs.end();
++cit){
string::value_type c= (*cit);
if(std::isalnum(c)||c=='.'||c=='_'||c=='-'||c=='~'){
oss << c;
continue;
}
oss << std::uppercase;
oss << '%' << std::setw(2) << size_t(c);
oss << std::nouppercase;
}
return oss.str();
}

string postElasticsearch(const string &search, const string &query){
http::client client;
http::client::request request(search);   //如果请求方式为get时，创建请求的url为search和编码后的query拼接得到的
request << network::header("Content-Length",
boost::lexical_cast<string>(query.size()));
//request << network::header("Content-Type","application/x-www-form-urlencoded;");
request << network::header("Connection","close");
request << network::body(query);

http::client::response response=client.post(request);  //请求采用的是post
return body(response);
}

std::pair<string, Json::Value>
parseJsonFirst(const string &value){
Json::Value root;
Json::Reader reader;
reader.parse(value, root);
string _scroll_id=root["_scroll_id"].asString();
Json::Value hits=root["hits"];
return std::make_pair(_scroll_id, hits);
}

int main(int argc, char** argv){
const string search=
"http://eshost:9200/your_index/_search?scroll=1m&search_type=scan";
const string scroll=
"http://eshost:9200/_search/scroll?scroll=1m";
const string query="{\"query\":{\"match_all\":{}}}";   //查询语句
string res=postElasticsearch(search, query);
//cout << res << endl;
SJPair rootpair=parseJsonFirst(res);
string _scroll_id(rootpair.first);
size_t total=(size_t)rootpair.second["total"].asInt();
//cout << _scroll_id << endl;
int Tsize = total/(10*8)+1;
StrVec vec;
for(int i=0;i<Tsize;++i){
string test=postElasticsearch(scroll, _scroll_id);
SJPair sonpair=parseJsonFirst(test);
//cout << sonpair.first << endl;
Json::Value hits=sonpair.second["hits"];
//cout << hits.size() << endl;
for(auto it=hits.begin();it!=hits.end();++it){
Json::Value temp=*it;
string _id=temp["_id"].asString();
Json::Value _source=temp["_source"];
string cell=_source["Clue_Entry_Cellphone"].asString();
//cout << _id <<" " << cell << endl;
if(cell.length()>0)
vec.push_back(cell);
}
_scroll_id=sonpair.first;
if(vec.size()>1000)
break;
}
cout << total << endl;
cout << Tsize << endl;
cout << vec.size() << endl;
}
}

为了使用 scroll，初始搜索请求应该在查询中指定 scroll 参数，这可以告诉 Elasticsearch 需要保持搜索的上下文环境多久（参考Keeping the search context alive），如

?scroll=1m

。

curl -XGET 'localhost:9200/twitter/tweet/_search?scroll=1m' -d '
{
"query": {
"match" : {
"title" : "elasticsearch"
}
}
}
'

使用上面的请求返回的结果中包含一个

scroll_id

，这个 ID 可以被传递给

scroll

API 来检索下一个批次的结果。

curl -XGET  'localhost:9200/_search/scroll'  -d'
{
"scroll" : "1m",
"scroll_id" : "c2Nhbjs2OzM0NDg1ODpzRlBLc0FXNlNyNm5JWUc1"
}
'

GET

或者

POST

可以使用

URL不应该包含

index

或者

type

名字——这些都指定在了原始的

search

请求中。

scroll

参数告诉 Elasticsearch 保持搜索的上下文等待另一个

1m

scroll_id

参数

每次对

scroll

API 的调用返回了结果的下一个批次知道没有更多的结果返回，也就是直到

hits

数组空了。

为了向前兼容，

scroll_id

和

scroll

可以放在查询字符串中传递。

scroll_id

则可以在请求体中传递。

curl -XGET 'localhost:9200/_search/scroll?scroll=1m' -d 'c2Nhbjs2OzM0NDg1ODpzRlBLc0FXNlNyNm5JWUc1'

注意：初始搜索请求和每个后续滚动请求返回一个新的

_scroll_id

——只有最近的

_scroll_id

才能被使用。

如果请求指定了聚合（aggregation），仅仅初始搜索响应才会包含聚合结果。

使用 scroll-scan 的高效滚动

使用

from and size

的深度分页，比如说

?size=10&from=10000

是非常低效的，因为

100,000

排序的结果必须从每个分片上取出并重新排序最后返回

10

条。这个过程需要对每个请求页重复。

scroll

API 保持了哪些结果已经返回的记录，所以能更加高效地返回排序的结果。但是，按照默认设定排序结果仍然需要代价。

一般来说，你仅仅想要找到结果，不关心顺序。你可以通过组合

scroll

和

scan

来关闭任何打分或者排序，以最高效的方式返回结果。你需要做的就是将

search_type=scan

加入到查询的字符串中：

curl -XGET 'localhost:9200/twitter/tweet/_search?scroll=1m&search_type=scan' -d '
{
"query": {
"match" : {
"title" : "elasticsearch"
}
}
}
'

设置

search_type

为

scan

可以关闭打分，让滚动更加高效。