ElasticSearch
与 Elastic Search 交互
| 限定名称 | 类型 | 版本 |
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
目前无法通过证书查询 Elastic 8，只能通过配置 |
示例
call apoc.es.post("localhost","tweets","users",null,{name:"Chris"})call apoc.es.put("localhost","tweets","users","1",null,{name:"Chris"})call apoc.es.get("localhost","tweets","users","1",null,null)call apoc.es.stats("localhost")call apoc.es.delete("localhost","indexName","typeName","idName")
分页
要使用 Elasticsearch 的分页功能，必须遵循以下步骤：
-
调用 apoc.es.query 获取第一块数据，并同时获取 scroll_id（以启用分页）。
-
使用前 N 条记录执行你的 merge/create 等操作。
-
使用 range(start,end,step) 函数重复调用，以获取剩余的所有数据块直到结束。例如，如果你有 1000 个文档，且每次请求获取 10 个文档，你可以执行 range(11,1000,10)。从 11 开始是因为前 10 个文档已经处理过了。如果你不知道准确的上限（文档总数），可以设置一个大于实际总数的数字。
-
重复调用的第二个函数是 apoc.es.get。记得将 scroll_id 设置为参数。
-
然后像处理第一块数据一样处理后续每一块数据的结果。
以下是一个示例：
// It's important to create an index to improve performance
CREATE INDEX FOR (n:Document) ON (n.id)
// First query: get first chunk of data + the scroll_id for pagination
CALL apoc.es.query('localhost','test-index','test-type','name:Neo4j&size=1&scroll=5m',null) yield value with value._scroll_id as scrollId, value.hits.hits as hits
// Do something with hits
UNWIND hits as hit
// Here we simply create a document and a relation to a company
MERGE (doc:Document {id: hit._id, description: hit._source.description, name: hit._source.name})
MERGE (company:Company {name: hit._source.company})
MERGE (doc)-[:IS_FROM]->(company)
// Then call for the other docs and use the scrollId value from previous query
// Use a range to count our chunk of data (i.e. i want to get chunks from 2 to 10)
WITH range(2,10,1) as list, scrollId
UNWIND list as count
CALL apoc.es.get("localhost","_search","scroll",null,{scroll:"5m",scroll_id:scrollId},null) yield value with value._scoll_id as scrollId, value.hits.hits as nextHits
// Again, do something with hits
UNWIND nextHits as hit
MERGE (doc:Document {id: hit._id, description: hit._source.description, name: hit._source.name})
MERGE (company:Company {name: hit._source.company})
MERGE (doc)-[:IS_FROM]->(company) return scrollId, doc, company该示例在拥有 16GB 内存的 MacBook Pro 上进行了测试。将 20000 个文档从 ES 加载到 Neo4j（每次请求 100 个文档）耗时 1 分钟。
常规结构与参数
call apoc.es.post(host-or-key,index-or-null,type-or-null,id-or-null,query-or-null,payload-or-null,$config) yield value
// GET/PUT/POST url/index/type/id?query -d payloadhost-or-key 参数
该参数可以是：
-
host
-
host:port
-
username:password@host:port
例如，使用 apoc.es.stats 时，我们可以执行：
CALL apoc.es.stats('http://username:password@host:port')此外，它可以是在 apoc.conf 中查询的条目：
-
查找 apoc.es.url
-
查找 apoc.es.host
这比作为第一个参数的直接字符串主机或 URL 具有更高的优先级。
例如，当 apoc.conf 如下所示时：
apoc.es.url=http://username:password@host:port或像这样：
apoc.es.host=username:password@host:port我们可以通过将第一个参数设为 null 来连接到 elastic。
例如，使用 apoc.es.stats 时，我们可以执行：
CALL apoc.es.stats(null)此外，它可以是在 apoc.conf 中查询的条目，其中 <key> 必须放在第一个参数中：
-
通过键查找 apoc.es.<key>.url
-
通过键查找 apoc.es.<key>.host
例如，当 apoc.conf 如下所示时：
apoc.es.custom.url=http://username:password@host:port或者像这样：
apoc.es.custom.host=username:password@host:port我们可以通过将第一个参数设为 null 来连接到 elastic。
例如，使用 apoc.es.stats 时，我们可以执行：
CALL apoc.es.stats('custom')config 参数
Config 可以是可选的 map，它可以包含以下条目：
| 名称 (name) | type | 默认 | description（描述） |
|---|---|---|---|
headers |
|
{ |
包含一个用于添加（或替换）默认头的头部映射。APOC 需要 |
version |
|
|
可以是 |
例如，使用 apoc.es.stats 时，我们可以执行：
CALL apoc.es.stats('custom', { headers: {Authorization: "Basic <Base64Token>"} })用于使用 基本身份验证 (Basic authentication) 并创建以下 HTTP 头部：
Authorization: Basic <Base64Token>
method: GET
Content-Type: application/jsonElastic 8 中的某些 API 可以由过程直接调用，无需 {version: 'EIGHT'} 配置，例如 apoc.es.stats，但对于许多 API，必须设置它才能正确处理端点，例如 apoc.es.query。
| 过程 | 使用版本: DEFAULT |
使用版本: EIGHT |
|---|---|---|
|
<host>/_stats |
同 |
|
<host>/<index param>/<type param>/_stats?<query param> |
<host>/<index param>/_stats?<query param> |
|
|
同 |
其他 |
|
类型参数通常是一个下划线字符串，表示 API 的类型，例如 |
例如，使用 apoc.es.query，我们可以执行搜索 API。
CALL apoc.es.query(<$host>, <$index>, <$type>, 'q=name:Neo4j', null, { version: 'EIGHT' })通过 Update API 更新 Elastic 8 中的文档。
CALL apoc.es.put($host,'<indexName>','_doc','<idName>','refresh=true',{name: 'foo'}, {version: 'EIGHT'})在 Elastic 8 中调用 创建索引 API。
CALL apoc.es.put($host,'<indexName>', null, null, null, null, { version: 'EIGHT' })倒数排名融合 (RRF)
可以使用 Neo4j 通过 ES 执行 RRF。有关详细信息，请阅读 官方文档。请注意，此 API 自 Elastic 8.14.x 版本起受支持。
以下是 Neo4j 与 ES 结合使用的示例。
步骤 1 - 创建映射
CALL apoc.es.put($host, 'example-index', null, null, null,
{
"mappings": {
"properties": {
"text": {
"type": "text"
},
"vector": {
"type": "dense_vector",
"dims": 1,
"index": true,
"similarity": "l2_norm"
},
"integer": {
"type": "integer"
}
}
}
}, $config)步骤 2 - 存入文档
CALL apoc.es.put($host, 'example-index/_doc/1', null, null, null,
{
"text" : "rrf",
"vector" : [5],
"integer": 1
}, $config)
CALL apoc.es.put($host, 'example-index/_doc/2', null, null, null,
{
"text" : "rrf rrf",
"vector" : [4],
"integer": 2
}, $config)
CALL apoc.es.put($host, 'example-index/_doc/3', null, null, null,
{
"text" : "rrf rrf rrf",
"vector" : [3],
"integer": 1
}, $config)
CALL apoc.es.put($host, 'example-index/_doc/4', null, null, null,
{
"text" : "rrf rrf rrf rrf",
"integer": 2
}, $config)
CALL apoc.es.put($host, 'example-index/_doc/5', null, null, null,
{
"vector" : [0],
"integer": 1
}, $config)步骤 4 - 使用 RRF 检索器执行搜索
CALL apoc.es.getRaw($host,'example-index/_search',
{
"retriever": {
"rrf": {
"retrievers": [
{
"standard": {
"query": {
"term": {
"text": "rrf"
}
}
}
},
{
"knn": {
"field": "vector",
"query_vector": [3],
"k": 5,
"num_candidates": 5
}
}
],
"window_size": 5,
"rank_constant": 1
}
},
"size": 3,
"aggs": {
"int_count": {
"terms": {
"field": "integer"
}
}
}
}
,$config) yield value