package org.fanli.elastic;
|
|
import static org.junit.Assert.assertNull;
|
|
import java.io.IOException;
|
import java.util.Collections;
|
import java.util.List;
|
import java.util.Map;
|
|
import org.apache.http.HttpHost;
|
import org.elasticsearch.action.bulk.BulkItemResponse;
|
import org.elasticsearch.action.get.GetResponse;
|
import org.elasticsearch.action.get.MultiGetItemResponse;
|
import org.elasticsearch.action.get.MultiGetRequest;
|
import org.elasticsearch.action.get.MultiGetResponse;
|
import org.elasticsearch.action.support.IndicesOptions;
|
import org.elasticsearch.client.RequestOptions;
|
import org.elasticsearch.client.RestClient;
|
import org.elasticsearch.client.RestHighLevelClient;
|
import org.elasticsearch.client.RethrottleRequest;
|
import org.elasticsearch.client.core.MultiTermVectorsRequest;
|
import org.elasticsearch.client.core.MultiTermVectorsResponse;
|
import org.elasticsearch.client.core.TermVectorsRequest;
|
import org.elasticsearch.client.core.TermVectorsResponse;
|
import org.elasticsearch.common.Strings;
|
import org.elasticsearch.common.unit.TimeValue;
|
import org.elasticsearch.common.xcontent.XContentBuilder;
|
import org.elasticsearch.common.xcontent.XContentFactory;
|
import org.elasticsearch.index.VersionType;
|
import org.elasticsearch.index.query.TermQueryBuilder;
|
import org.elasticsearch.index.reindex.BulkByScrollResponse;
|
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
|
import org.elasticsearch.index.reindex.ReindexRequest;
|
import org.elasticsearch.index.reindex.ScrollableHitSource;
|
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
|
import org.elasticsearch.script.Script;
|
import org.elasticsearch.script.ScriptType;
|
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
|
import org.elasticsearch.search.sort.SortOrder;
|
import org.elasticsearch.tasks.TaskId;
|
|
public class QueryDOUtils {
|
|
private static String clusterName = "my-application";
|
private static String host = "192.168.1.200";
|
private static Integer port = 9200;
|
|
// 相当于数据库名称
|
public static String indexName = "shose";
|
|
// 初始化api客户端
|
public static RestHighLevelClient client = new RestHighLevelClient(
|
RestClient.builder(new HttpHost(host, port, "http")));
|
|
public static void MultiGetRequest() {
|
MultiGetRequest request = new MultiGetRequest();
|
// Index、Document id
|
request.add(new MultiGetRequest.Item("index", "example_id"));
|
request.add(new MultiGetRequest.Item("index", "another_id"));
|
|
/* 可选参数 */
|
// 禁用源检索，默认情况下启用
|
request.add(new MultiGetRequest.Item("index", "example_id")
|
.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE));
|
|
// 1、为特定字段配置源包含
|
String[] includes = new String[] { "foo", "*r" };
|
String[] excludes = Strings.EMPTY_ARRAY;
|
FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
|
request.add(new MultiGetRequest.Item("index", "example_id").fetchSourceContext(fetchSourceContext));
|
|
// 2、为特定字段配置源排除
|
String[] includes2 = Strings.EMPTY_ARRAY;
|
String[] excludes2 = new String[] { "foo", "*r" };
|
FetchSourceContext fetchSourceContext2 = new FetchSourceContext(true, includes2, excludes2);
|
request.add(new MultiGetRequest.Item("index", "example_id").fetchSourceContext(fetchSourceContext2));
|
|
// 配置特定存储字段的检索（要求将字段分别存储在映射中） 检索foo存储的字段（要求将字段单独存储在映射中）
|
request.add(new MultiGetRequest.Item("index", "example_id").storedFields("foo"));
|
try {
|
MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
|
MultiGetItemResponse item = response.getResponses()[0];
|
String value = item.getResponse().getField("foo").getValue();
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
|
request.add(new MultiGetRequest.Item("index", "with_routing").routing("some_routing"));
|
request.add(
|
new MultiGetRequest.Item("index", "with_version").versionType(VersionType.EXTERNAL).version(10123L));
|
|
// 偏好值
|
request.preference("some_preference");
|
// 将实时标志设置为false（true默认情况下）
|
request.realtime(false);
|
// 检索文档之前执行刷新（false默认情况下）
|
request.refresh(true);
|
|
try {
|
MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
|
|
MultiGetItemResponse firstItem = response.getResponses()[0];
|
// getFailure 返回null，因为没有失败。
|
assertNull(firstItem.getFailure());
|
|
GetResponse firstGet = firstItem.getResponse();
|
String index = firstItem.getIndex();
|
String id = firstItem.getId();
|
if (firstGet.isExists()) {
|
long version = firstGet.getVersion();
|
// 将该文档检索为 String
|
String sourceAsString = firstGet.getSourceAsString();
|
// 将该文档检索为 Map<String, Object>
|
Map<String, Object> sourceAsMap = firstGet.getSourceAsMap();
|
// 将该文档检索为 byte[]
|
byte[] sourceAsBytes = firstGet.getSourceAsBytes();
|
} else {
|
// 处理找不到文档的情况。请注意，尽管返回的响应具有404状态码，但返回的是有效值GetResponse，而不是引发异常。
|
// 这样的响应不包含任何源文档，并且其isExists方法返回false。
|
}
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
}
|
|
public static void queryUpdate() {
|
// 一个ReindexRequest可以用来从一个或多个索引文件复制到目标指数。
|
// 它要求在请求之前可能存在或可能不存在的现有源索引和目标索引。Reindex不会尝试设置目标索引。
|
// 它不会复制源索引的设置。您应该在运行_reindex操作之前设置目标索引，包括设置映射，分片计数，副本等。
|
ReindexRequest request = new ReindexRequest();
|
request.setSourceIndices("source1", "source2"); // 添加要复制的来源列表
|
request.setDestIndex("dest"); // 添加目标索引
|
|
// 。设置versionType为external将导致Elasticsearch从源保留版本，创建丢失的所有文档，并更新目标索引中比源索引中具有旧版本的任何文档。
|
request.setDestVersionType(VersionType.EXTERNAL);
|
// 设置opType为create将会导致_reindex仅在目标索引中创建丢失的文档。所有现有文档都将导致版本冲突。默认opType值为index。
|
request.setDestOpType("create");
|
// 默认情况下，版本冲突会中止该_reindex过程，但是您可以使用以下方法来计算它们：
|
request.setConflicts("proceed");
|
|
// 添加查询来限制文档。 仅复制字段user设置为kimchy
|
request.setSourceQuery(new TermQueryBuilder("user", "kimchy"));
|
// 限制已处理文档的数量maxDocs。
|
request.setMaxDocs(10);
|
// 默认情况下_reindex使用的批次为1000。您可以使用更改批次大小sourceBatchSize。
|
request.setSourceBatchSize(100);
|
|
// Reindex也可以通过指定来使用摄取功能pipeline。
|
request.setDestPipeline("my_pipeline");
|
|
// 如果要从源索引中获取一组特定的文档，则需要使用sort。如果可能，请选择更具选择性的查询，而不是maxDocs和排序。
|
request.addSortField("field1", SortOrder.DESC);
|
request.addSortField("field2", SortOrder.ASC);
|
|
// 支持script修改文档。它还允许您更改文档的元数据。
|
// setScriptlikes使用user 增大所有文档上的字段kimchy。
|
request.setScript(new Script(ScriptType.INLINE, "painless",
|
"if (ctx._source.user == 'kimchy') {ctx._source.likes++;}", Collections.emptyMap()));
|
|
// ReindexRequest支持从远程Elasticsearch集群重新索引。使用远程集群时，应在RemoteInfo对象内部指定查询，而不要使用setSourceQuery。
|
// 如果同时设置了远程信息和源查询，则会在请求期间导致验证错误。这样做的原因是远程Elasticsearch可能无法理解现代查询构建器构建的查询。
|
// 远程集群支持一直运行到Elasticsearch 0.90，此后查询语言已更改。达到旧版本时，以JSON手动编写查询会更安全。
|
// request.setRemoteInfo(new RemoteInfo("http", remoteHost, remotePort, null,
|
// new BytesArray(new MatchAllQueryBuilder().toString()), user, password, Collections.emptyMap(),
|
// new TimeValue(100, TimeUnit.MILLISECONDS), new TimeValue(100, TimeUnit.SECONDS)));
|
|
try {
|
BulkByScrollResponse bulkResponse = client.reindex(request, RequestOptions.DEFAULT);
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
}
|
|
public static void queryUpdate2() {
|
UpdateByQueryRequest request = new UpdateByQueryRequest("source1", "source2");
|
|
request.setConflicts("proceed");
|
request.setQuery(new TermQueryBuilder("user", "kimchy"));
|
request.setMaxDocs(10);
|
request.setBatchSize(100);
|
request.setPipeline("my_pipeline");
|
|
request.setScript(new Script(ScriptType.INLINE, "painless",
|
"if (ctx._source.user == 'kimchy') {ctx._source.likes++;}", Collections.emptyMap()));
|
|
request.setSlices(2);
|
|
request.setScroll(TimeValue.timeValueMinutes(10));
|
request.setRouting("=cat");
|
|
request.setTimeout(TimeValue.timeValueMinutes(2));
|
request.setRefresh(true);
|
request.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
|
try {
|
BulkByScrollResponse bulkResponse = client.updateByQuery(request, RequestOptions.DEFAULT);
|
|
// 获取总时间
|
TimeValue timeTaken = bulkResponse.getTook();
|
// 检查请求是否超时
|
boolean timedOut = bulkResponse.isTimedOut();
|
// 获取处理的文档总数
|
long totalDocs = bulkResponse.getTotal();
|
// 已更新的文档数
|
long updatedDocs = bulkResponse.getUpdated();
|
// 被删除的文档数
|
long deletedDocs = bulkResponse.getDeleted();
|
// 已执行的批次数
|
long batches = bulkResponse.getBatches();
|
// 跳过的文档数
|
long noops = bulkResponse.getNoops();
|
// 版本冲突数
|
long versionConflicts = bulkResponse.getVersionConflicts();
|
// 请求必须重试批量索引操作的次数
|
long bulkRetries = bulkResponse.getBulkRetries();
|
// 请求必须重试搜索操作的次数
|
long searchRetries = bulkResponse.getSearchRetries();
|
// 如果该请求当前处于睡眠状态，则该请求已进行自身限制的总时间不包括当前的限制时间
|
TimeValue throttledMillis = bulkResponse.getStatus().getThrottled();
|
// 当前节气门睡眠的剩余延迟；如果未睡眠，则为0
|
TimeValue throttledUntilMillis = bulkResponse.getStatus().getThrottledUntil();
|
// 搜索阶段失败
|
List<ScrollableHitSource.SearchFailure> searchFailures = bulkResponse.getSearchFailures();
|
// 大容量索引操作期间发生故障
|
List<BulkItemResponse.Failure> bulkFailures = bulkResponse.getBulkFailures();
|
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
}
|
|
// 按查询删除
|
public static void queryDelete() {
|
DeleteByQueryRequest request = new DeleteByQueryRequest("source1", "source2");
|
request.setConflicts("proceed");
|
request.setQuery(new TermQueryBuilder("user", "kimchy"));
|
request.setMaxDocs(10);
|
request.setBatchSize(100);
|
request.setSlices(2); // 设置要使用的切片数
|
request.setScroll(TimeValue.timeValueMinutes(10)); // 使用scroll参数来控制它使“搜索上下文”保持活动的时间。
|
request.setRouting("=cat");
|
request.setTimeout(TimeValue.timeValueMinutes(2)); // 等待通过查询请求执行删除的超时 TimeValue
|
request.setRefresh(true); // 通过查询调用删除后刷新索引
|
request.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN); // 设置索引选项
|
try {
|
BulkByScrollResponse bulkResponse = client.deleteByQuery(request, RequestOptions.DEFAULT);
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
}
|
|
public static void RethrottleRequest(TaskId taskId) {
|
// 可用于更改正在运行的重新索引，按查询更新或按查询删除任务的当前限制，或完全禁用该任务的限制。它需要更改任务的任务ID。
|
RethrottleRequest request = new RethrottleRequest(taskId);
|
|
RethrottleRequest request2 = new RethrottleRequest(taskId, 100.0f);
|
|
try {
|
// 执行重新索引重新调节请求
|
client.reindexRethrottle(request, RequestOptions.DEFAULT);
|
// 通过查询更新相同
|
client.updateByQueryRethrottle(request, RequestOptions.DEFAULT);
|
// 通过查询删除相同
|
client.deleteByQueryRethrottle(request, RequestOptions.DEFAULT);
|
} catch (Exception e) {
|
}
|
}
|
|
public static void main(String[] args) {
|
System.out.println("------------ 测试结束 -------------------------");
|
}
|
|
}
|
