ES2.x版本的javaAPI使用实例

前段时间做用户标签系统，使用的es2.2来存储查询。其中涉及到了es的javaAPI的使用。这里把笨小葱用到的一些函数分享一下。

首先是es的连接获取操作实例。

import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.sort.SortOrder;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
* Created by cc on 2016/4/4.
*/
public class Es_Client {
public static final String CLUSTER_NAME = "razor_es"; //实例名称
private static final String IP = "192.168.1.1";
private static final int PORT = 9300; //端口
//1.设置集群名称：默认是elasticsearch，并设置client.transport.sniff为true，使客户端嗅探整个集群状态，把集群中的其他机器IP加入到客户端
//对ES2.0有效
private static Settings settings = Settings
.settingsBuilder()
.put("cluster.name", CLUSTER_NAME)
.put("client.transport.sniff", true)
.build();

//创建私有对象
private static TransportClient client;

static {
try {
client = TransportClient.builder().settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), PORT));
} catch (UnknownHostException e) {
e.printStackTrace();
}
}

//取得实例
public static TransportClient getTransportClient() {
return client;
}

}


然后是一些函数操作：

import com.tesla.razor.dao.TagDAO;
import com.tesla.razor.util.Es_Client;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.sort.SortOrder;

import java.util.ArrayList;
import java.util.List;

public class TagDAOimpl implements TagDAO {

@Override
public JSONObject getUserInfoList(String productid, String startdate, String stopdate) throws Exception {

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.matchQuery("product_id", productid));
boolQueryBuilder.must(QueryBuilders.rangeQuery("localtime").from(startdate).to(stopdate));

SearchResponse response = Es_Client.getTransportClient().prepareSearch("razor_cd")
.setTypes("clientdata")
.setQuery(boolQueryBuilder)
/* .addAggregation(
AggregationBuilders.terms("agg").field("user_id").size(20)
)*/
.addAggregation(
AggregationBuilders.cardinality("agg1").field("user_id")
)
//.addAggregation(AggregationBuilders.dateRange("range1").field("localtime").addRange(startdate,stopdate))
.addSort("localtime", SortOrder.DESC)
.setSize(2000)
.execute()
.actionGet();

JSONObject object=new JSONObject();
object.put("total",response.getAggregations().getProperty("agg1.value").toString());

JSONArray arr = new JSONArray();
//取前20条不重复的userid
List<String> userIds=new ArrayList<String>();
SearchHit[] hits= response.getHits().getHits();
for(SearchHit hit : hits)
{
if(userIds.size()<20)
{
String userid = hit.getSource().get("user_id").toString();
int flag = 1;
for (String s : userIds) {
if (s.equals(userid)) {
flag = 0;
break;
}
}
if (flag == 1)//flag为1表示还没有记录该用户
{
userIds.add(userid);
arr.add(hit.getSourceAsString());
}

}
}

object.put("userList",arr);
//System.out.println(object);
return object;
}

@Override
public JSONObject getUserInfoListById(String productid, String startdate, String stopdate,String userid) throws Exception {

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.matchQuery("product_id", productid));
boolQueryBuilder.must(QueryBuilders.regexpQuery("user_id", ".*" + userid + ".*"));
boolQueryBuilder.must(QueryBuilders.rangeQuery("localtime").from(startdate).to(stopdate));

SearchResponse response = Es_Client.getTransportClient().prepareSearch("razor_cd")
.setTypes("clientdata")
.setQuery(boolQueryBuilder) // Filter
/*.addAggregation(
AggregationBuilders.terms("agg").field("user_id").size(20)
)*/
.addAggregation(
AggregationBuilders.cardinality("agg1").field("user_id")
)
.addSort("localtime",SortOrder.DESC)
.setSize(2000)
.execute()
.actionGet();

JSONObject object=new JSONObject();
object.put("total",response.getAggregations().getProperty("agg1.value").toString());

JSONArray arr = new JSONArray();
//取前20条不重复的userid
List<String> userIds=new ArrayList<String>();
SearchHit[] hits= response.getHits().getHits();
for(SearchHit hit : hits)
{
if(userIds.size()<20)
{
String user_id = hit.getSource().get("user_id").toString();
int flag = 1;
for (String s : userIds) {
if (s.equals(user_id)) {
flag = 0;
break;
}
}
if (flag == 1)//flag为1表示还没有记录该用户
{
userIds.add(user_id);
arr.add(hit.getSourceAsString());
}

}
}

object.put("userList",arr);

//System.out.println(object);
return object;
}

@Override
public JSONObject getUserInfoListByTags(String productid, String tags) throws Exception {
String[] tags_arr=tags.split(";");
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.matchQuery("product_id", productid));
for(String tag : tags_arr)
{
if(tag.contains("c_channel_"))
{
boolQueryBuilder.must(QueryBuilders.matchQuery("channel_id", tag.substring(10)));
}
else if(tag.contains("c_provinc_"))
{
boolQueryBuilder.must(QueryBuilders.matchQuery("region", tag.substring(10)));
}
else
{
boolQueryBuilder.must(QueryBuilders.matchQuery("app_tags.tag_name", tag));
}

}

SearchResponse response = Es_Client.getTransportClient().prepareSearch("razor_tag")
.setTypes("user_tag")
.setQuery(boolQueryBuilder)
//.addField("user_id")
.setSize(20)
.execute()
.actionGet();

//System.out.println(response.toString());

//根据查询出的用户id去clientdata中二次查询详细信息
JSONObject object=new JSONObject();
object.put("total",response.getHits().getTotalHits());

JSONArray arr = new JSONArray();
//取前20条不重复的userid
List<String> userIds=new ArrayList<String>();
SearchHit[] hits= response.getHits().getHits();
for(SearchHit hit : hits)
{
<span style="white-space:pre"> </span>......<span style="white-space:pre"> </span>
}

return object;
}

@Override
public String getUserDetailTags(String productid, String userid) throws Exception {

GetResponse usertag= Es_Client.getTransportClient().prepareGet("razor_tag", "user_tag", userid + "|" + productid).get();
return usertag.getSourceAsString();
}

@Override
public JSONArray getUserDetail(String productid, String startdate, String stopdate,String userid) throws Exception {

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.matchQuery("product_id", productid));
boolQueryBuilder.must(QueryBuilders.regexpQuery("user_id", userid));
boolQueryBuilder.must(QueryBuilders.rangeQuery("localtime").from(startdate).to(stopdate));

SearchResponse response = Es_Client.getTransportClient().prepareSearch("razor_cd")
.setTypes("clientdata")
.setQuery(boolQueryBuilder) // Filter
.addSort("localtime", SortOrder.DESC)
.setSize(20)
.execute()
.actionGet();

//Map<String,List<String>> map=new HashMap<String, List<String>>();
JSONArray arr = new JSONArray();

SearchHit[] hits= response.getHits().getHits();
String tmp="";
int flag=1;

for(SearchHit hit : hits)
{
System.out.println(hit.getSourceAsString());
JSONObject jsonObject = new JSONObject();
String sessionid=hit.getSource().get("session_id").toString();
if(flag==1) //表示取第一条记录
{
tmp=sessionid;
flag=0;
}
else { //取第1条之后的记录
if(tmp.equals(sessionid)) //重复的sessionid，直接结束，进入下次循环
{
continue;
}
else {
tmp=sessionid;
}
}

//1.添本次session的clientdata数据
jsonObject.put("clientdata",hit.getSourceAsString());
//2.添本次session的usinglog
GetResponse usinglog= Es_Client.getTransportClient().prepareGet("razor_usinglog", "usinglog", userid + "|" + sessionid).get();
jsonObject.put("usinglog", usinglog.getSourceAsString());
//3.添本次session的event
GetResponse event= Es_Client.getTransportClient().prepareGet("razor_event", "event", userid + "|" + sessionid).get();
jsonObject.put("event",event.getSourceAsString());

arr.add(jsonObject);
}

return arr;
}

@Override
public String getTagTop(String productid) throws Exception {
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.matchQuery("product_id", productid));
//boolQueryBuilder.must(QueryBuilders.rangeQuery("add_time").from(startdate).to(stopdate));

SearchResponse response = Es_Client.getTransportClient().prepareSearch("razor_tag")
.setTypes("user_tag")
.setQuery(boolQueryBuilder)
.addAggregation(
AggregationBuilders.terms("agg").field("app_tags.tag_name").size(20)
)
.addAggregation(
AggregationBuilders.cardinality("agg1").field("app_tags.tag_name")
)
.setSize(0)
.execute()
.actionGet();

return response.toString();
}

}