1.基于elasticsearch的KBQA实现及示例

http://openkg.cn/tool/elasticsearch-kbqa

2.ElasticSearch 处理自然语言流程

https://www.cnblogs.com/yisany/p/10292049.html

3.一周热点：新冠疫苗加强针开打，你关心的五个问题都有回应了

https://www.thepaper.cn/newsDetail_forward_15049026

• 编辑：bsll

• 归档：https://ela.st/cn-daily-all

• 订阅：https://ela.st/cn-daily-sub

• 沙龙：https://ela.st/cn-meetup

ES灾备技术调研

早就听说极限网关了，最近有个ES要做灾备，就搜集了下技术方案：

1. 应用双写两个集群

2. 跨集群复制CCR

3. 极限网关

三个方案的优缺点我也简单整理了下：

1. 优点：自主研发，想怎样就怎样；缺点：难，水平、时间都是成本

2. 优点：实施简单；缺点：需要license

3. 优点：部署灵活，功能齐全；缺点：不熟悉，没用过

貌似方案 3，比较理想。不熟悉，那就多用用，测试测试找找感觉。

测试环境 笔记本

网关下载

先下载极限网关：http://release.elasticsearch.cn/gateway/snapshot/

解压后，可看到有几个预定义的配置，方便大家快速上车。

网关配置

这次是做灾备，应该就是在 dc-replica.yml的基础上改比较妥当，配置还是比较多的 300多行，很多都是为了保证两个集群数据一致性，解耦啥的做的判断和写队列之类的设置。我只修改了 elasticsearch资源的定义信息，还有网关也可选择tls，不用手工生成证书，非常方便。

贴下我修改的地方

启动网关

两个集群都是 available ，说明连接应该ok

测试场景：围绕主备集群之间数据复制展开

1. 主备集群间复制创建、写入、删除索引操作

目的：对网关（A集群）创建索引test，对test索引写文档，能自动同步到B集群

初始情况A、B无test

手工put test索引到网关

用loadgen写点文档进去吧，loadgen直接写网关

都10000条写入成功，如果查看是0的话，先refresh下索引

看下文档

简单校验一下

感觉没问题，删点文档看看，把那23个删了吧

再看看

更新524个文档看看

增、删、改 完事

2. 主集群故障的处理和恢复

目的：主集群故障后，网关如何处理收到的请求：读、写

Kill 掉 A 集群，模拟主集群故障

网关直接有报错说连不上A集群

此时对网关查询、写入数据都会报错

重新启动主集群后，一切正常

3. 备集群故障的处理和恢复

目的：备集群故障，读、写主集群不受影响，备集群恢复后，自动追数据直到两边数据一致

Kill 掉B集群，网关报错，B集群无法连接

对A集群进行写入一个新文档，并查询下

启动B集群，查看，已自动添加了新的文档

4. 主集群容灾切换

目的：主集群故障，短时间不可用，网关切换指向备集群作为主集群，A集群作为备集群；待A集群恢复后，自动追数据

KILL A 集群 模拟A集群故障

Kill 网关，切换配置：

启动网关，B是主了，available

写B集群，删除前面创建的 test 索引，新建test2索引，并插入文档

启动A集群，检查数据，有test2索引，且文档一致

5. 主集群故障恢复后，网关回切

目的：假设A集群已经完全恢复，网关回切，A集群为主，B集群为备

Kill网关，恢复配置，再启动网关

再次测试写入test2 检查能否同步成功

写网关

查看A集群

查看B集群

两个集群数据一致，主备角色也恢复到最初了

**测试环境变更

中间工作比较忙，中断了测试，也就十来天吧

不成想 网关更新了，我就下了新版本的gateway和ES-7.15.1 继续捣鼓

环境 笔记本

网关不光版本升级了，配置文件也稍稍有些变化，新版本示例配置如下

本着不会就猜的原则，我还是能做到见名之知意的

我仅仅对 cross-cluster-replication.yml 做了简单的修改，就是和前面一样

6. ES由代理暴露的能否正常工作

目的：检验，比如由nginx代理ES的时候，网关能否正常工作

使用 nginx 代理后端的ES

80-->primary

81-->backup

测试下反向代理工作是否正常

启动网关，网关里elasticsearch资源指向nginx，没错就改了下 url

开始测试，增删改查 once more，还是对着网关操作

创建索引

PUT 文档

抄家伙

看看数量

简单对比下

直接通过代理查询，检验数据

没问题，测试删除

删除单条

批量删除

看看结果

删除索引，并确认真的删除

关调backup集群，然后primary集群创建索引后，再启动backup集群，看能否自动追平

创建test2索引

顺手put个文档

Primary上查看

OK 启动backup

妈的 奇迹

7. 请求压缩功能如何

目的：看看客户端和网关分别在不同的请求压缩配置下，是否工作正常

在网关的配置中，看到了 compress 压缩，也设计几个场景测试下

网关配置里面有个压缩的配置，意思是消费线程往ES发送请求的时候是否启用gzip压缩。

默认是 true

7.1 用压缩的方式写网关，但网关不启用压缩

我们把网关配置中的true改成false

启动网关

开始写数据，写的过程中，随机停止backup集群，等待一会儿后再启动

网关可看到 backup集群，从不可用到可用

看看网关的队列情况，这时候backup队列应该堆积了不少请求

等待队列被消费完后，我们对比数据。

简单看看文档条数

用网关对比下数据看看

妥妥的一致

7.2 用压缩的方式写网关，网关也启用压缩

就把前面网关贴图的false改成true

启动网关

压缩写入

写入过程中，随机重启backup集群

网关日志

等待网关backup队列消费完毕，后对比数据

没得问题，下一场

7.3 用不压缩的方式写网关，网关也不启用压缩

网关配置compress: false，后启动网关

不压缩请求

中途随机重启backup集群

队列消费完后对比数据

数据一致

7.4 用不压缩的方式写网关，网关backup队列的消费线程使用压缩

写入过程中，停止backup集群，稍等一会儿后启动backup集群

观察网关的backup 队列情况

等消费完之后，我们对比数据

数量一致

用网关来对比下两个集群的test索引的数据

至此，关于跨集群复制的场景测试可以告一段落了，功能大家也看到了确实强大，部署也很灵活，对环境无侵入，直接把应用连接ES的地址、端口改下就行。或者网关直接监听9200端口，替换原有的协调节点，应用无感知。

此外，极限网关还有很多别的使用场景和激动人心的功能，像在线修改查询、请求限流、请求流量分析，这都是保障ES集群稳定运行的手段。

看到一款国产软件如此强大很欣慰，很激动。希望各位感兴趣的小伙伴也下载测试下自己的场景，我们ES圈（论坛）多交流交流。让我也看看你们的场景和测试是怎样的。

最后一个图，我也不知道咋回事，请忽略。

1.基于rust实现的es-dsl库

https://github.com/vinted/elasticsearch-dsl-rs

2.使用es进行url搜索

https://tech.trivago.com/2020/02/11/better-url-search-with-elasticsearch/

3.一周热点：央视网评｜李云迪跌落神坛完全是自作孽

https://baijiahao.baidu.com/s?id=1714245507501328373&wfr=spider&for=pc

• 编辑：bsll

• 归档：https://ela.st/cn-daily-all

• 订阅：https://ela.st/cn-daily-sub

• 沙龙：https://ela.st/cn-meetup

使用场景

如果你的业务需要用到有多个集群，并且版本还不一样，是不是管理起来很麻烦，如果能够通过一个 API 来进行查询就方便了，聪明的你相信已经想到了 CCS，没错用 CCS 可以实现跨集群的查询，不过 Elasticsearch 提供的 CCS 对版本有一点的限制，并且需要提前做好 mTLS，也就是需要提前配置好两个集群之间的证书互信，这个免不了要重启维护，好像有点麻烦，那么问题来咯，有没有更好的方案呢？

😁 有办法，今天我就给大家介绍一个基于极限网关的方案，极限网关的网址：http://极限网关.com/。

假设现在有两个集群，一个集群是 v2.4.6，有不少业务数据，舍不得删，里面有很多好东西 ：）还有一个集群是 v7.14.0，版本还算比较新，业务正在做的一个新的试点，没什么好东西，但是也得用 :(，现在老板们的的需求是希望通过在一个统一的接口就能访问这些数据，程序员懒得很，懂得都懂。

集群信息

• v2.4.6 集群的访问入口地址：192.168.3.188:9202
• v7.14.0 集群的访问入口地址：192.168.3.188:9206

这两个集群都是 http 协议的。

实现步骤

今天用到的是极限网关的 switch 过滤器：https://极限网关.com/docs/references/filters/switch/

网关下载下来就两个文件，一个主程序，一个配置文件，记得下载对应操作系统的包。

定义两个集群资源

elasticsearch:
  - name: v2
    enabled: true
    endpoint: http://192.168.3.188:9202
  - name: v7
    enabled: true
    endpoint: http://192.168.3.188:9206

上面定义了两个集群，分别命名为 v2 和 v7，待会会用到这些资源。

定义一个服务入口

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 1000
    network:
      binding: 0.0.0.0:8000
    tls:
      enabled: true

这里定义了一个名为 my_es_entry 的资源入口，并引用了一个名为 my_router 的请求转发路由，同时绑定了网卡的 0.0.0.0:8000 也就是所有本地网卡监听 IP 的 8000 端口，访问任意 IP 的 8000 端口就能访问到这个网关了。

另外老板也说了，Elasticsearch 用 HTTP 协议简直就是裸奔，通过这里开启 tls，可以让网关对外提供的是 HTTPS 协议，这样用户连接的 Elasticsearch 服务就自带 buffer 了，后端的 es 集群和网关直接可以做好网络流量隔离，集群不用动，简直完美。

为什么定义 TLS 不用指定证书，好用的软件不需要这么麻烦，就这样，不解释。

最后，通过设置 max_concurrency 为 1000，限制下并发数，避免野猴子把我们的后端的 Elasticsearch 给压挂了。

定义一个请求路由

router:
  - name: my_router
    default_flow: cross-cluster-search

这里的名称 my_router 就是表示上面的服务入口的router 参数指定的值。

另外设置一个 default_flow 来将所有的请求都转发给一个名为 cross-cluster-search 的请求处理流程，还没定义，别急，马上。

定义请求处理流程

来啦，来啦，先定义两个 flow，如下，分别名为 v2-flow 和 v7-flow，每节配置的 filter 定义了一系列过滤器，用来对请求进行处理，这里就用了一个 elasticsearch 过滤器，也就是转发请求给指定的 Elasticsearch 后端服务器，了否？

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7

然后，在定义额外一个名为 cross-cluster-search 的 flow，如下：

  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow

这个 flow 就是通过请求的路径的前缀来进行路由的过滤器，如果是 v2:开头的请求，则转发给 v2-flow 继续处理，如果是 v7: 开头的请求，则转发给 v7-flow 来处理，使用的用法和 CCS 是一样的。so easy！

对了，那是不是每个请求都需要加前缀啊，费事啊，没事，在这个 cross-cluster-search 的 filter 最后再加上一个 elasticsearch filter，前面前缀匹配不上的都会走它，假设默认都走 v7，最后完整的 flow 配置如下：

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7
  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow
      - elasticsearch:
          elasticsearch: v7              

然后就没有然后了，因为就配置这些就行了。

启动网关

假设配置文件的路径为 sample-configs/cross-cluster-search.yml，运行如下命令：

➜  gateway git:(master) ✗ ./bin/gateway -config sample-configs/cross-cluster-search.yml
   ___   _   _____  __  __    __  _       
  / _ \ /_\ /__   \/__\/ / /\ \ \/_\ /\_/\
 / /_\///_\\  / /\/_\  \ \/  \/ //_\\\_ _/
/ /_\\/  _  \/ / //__   \  /\  /  _  \/ \ 
\____/\_/ \_/\/  \__/    \/  \/\_/ \_/\_/ 

[GATEWAY] A light-weight, powerful and high-performance elasticsearch gateway.
[GATEWAY] 1.0.0_SNAPSHOT, 2021-10-15 16:25:56, 3d0a1cd
[10-16 11:00:52] [INF] [app.go:228] initializing gateway.
[10-16 11:00:52] [INF] [instance.go:24] workspace: data/gateway/nodes/0
[10-16 11:00:52] [INF] [api.go:260] api listen at: http://0.0.0.0:2900
[10-16 11:00:52] [INF] [reverseproxy.go:257] elasticsearch [v7] hosts: [] => [192.168.3.188:9206]
[10-16 11:00:52] [INF] [entry.go:225] auto generating cert files
[10-16 11:00:52] [INF] [actions.go:223] elasticsearch [v2] is available
[10-16 11:00:52] [INF] [actions.go:223] elasticsearch [v7] is available
[10-16 11:00:53] [INF] [entry.go:296] entry [my_es_entry] listen at: https://0.0.0.0:8000
[10-16 11:00:53] [INF] [app.go:309] gateway is running now.

可以看到网关输出了启动成功的日志，网关服务监听在 https://0.0.0.0:8000 。

试试访问网关

直接访问网关的 8000 端口，因为是网关自签的证书，加上 -k 来跳过证书的校验，如下：

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000  
{
  "name" : "LENOVO",
  "cluster_name" : "es-v7140",
  "cluster_uuid" : "npWjpIZmS8iP_p3GK01-xg",
  "version" : {
    "number" : "7.14.0",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1",
    "build_date" : "2021-07-29T20:49:32.864135063Z",
    "build_snapshot" : false,
    "lucene_version" : "8.9.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

正如前面配置所配置的一样，默认请求访问的就是 v7 集群。

访问 v2 集群

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:/    
{
  "name" : "Solomon O'Sullivan",
  "cluster_name" : "es-v246",
  "cluster_uuid" : "cqlpjByvQVWDAv6VvRwPAw",
  "version" : {
    "number" : "2.4.6",
    "build_hash" : "5376dca9f70f3abef96a77f4bb22720ace8240fd",
    "build_timestamp" : "2017-07-18T12:17:44Z",
    "build_snapshot" : false,
    "lucene_version" : "5.5.4"
  },
  "tagline" : "You Know, for Search"
}

查看集群信息：

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:_cluster/health\?pretty
{
  "cluster_name" : "es-v246",
  "status" : "yellow",
  "timed_out" : false,
  "number_of_nodes" : 1,
  "number_of_data_nodes" : 1,
  "active_primary_shards" : 5,
  "active_shards" : 5,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 5,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0,
  "task_max_waiting_in_queue_millis" : 0,
  "active_shards_percent_as_number" : 50.0
}

插入一条文档：

➜  loadgen git:(master) ✗ curl-json -k   https://localhost:8000/v2:medcl/doc/1 -d '{"name":"hello world"}'
{"_index":"medcl","_type":"doc","_id":"1","_version":1,"_shards":{"total":2,"successful":1,"failed":0},"created":true}%  

执行一个查询

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:medcl/_search\?q\=name:hello                
{"took":78,"timed_out":false,"_shards":{"total":5,"successful":5,"failed":0},"hits":{"total":1,"max_score":0.19178301,"hits":[{"_index":"medcl","_type":"doc","_id":"1","_score":0.19178301,"_source":{"name":"hello world"}}]}}% 

可以看到，所有的请求，不管是集群的操作，还是索引的增删改查都可以，而 Elasticsearch 自带的 CCS 是只读的，只能进行查询。

访问 v7 集群

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v7:/
{
  "name" : "LENOVO",
  "cluster_name" : "es-v7140",
  "cluster_uuid" : "npWjpIZmS8iP_p3GK01-xg",
  "version" : {
    "number" : "7.14.0",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1",
    "build_date" : "2021-07-29T20:49:32.864135063Z",
    "build_snapshot" : false,
    "lucene_version" : "8.9.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

Kibana 里面访问

完全没问题，有图有真相：

其他操作也类似，就不重复了。

完整的配置

path.data: data
path.logs: log

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: 0.0.0.0:8000
    tls:
      enabled: true

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7
  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow
      - elasticsearch:
          elasticsearch: v7

router:
  - name: my_router
    default_flow: cross-cluster-search

elasticsearch:
  - name: v2
    enabled: true
    endpoint: http://192.168.3.188:9202
  - name: v7
    enabled: true
    endpoint: http://192.168.3.188:9206

小结

好了，今天给大家分享的如何使用极限网关来进行 Elasticsearch 跨集群跨版本的操作就到这里了，希望大家周末玩的开心。😁

通过python脚本迁移ES的template模板

通过python脚本迁移ES的template模板，从192.168.0.1 迁移到 192.168.0.2

import base64
import json

import requests

def putTemplate(templateName, templateDslJson):
    print("{0} 索引模板正在迁移中".format(templateName))

    res = requests.put("http://192.168.0.2:9200/_template/{0}".format(templateName), json=templateDslJson)
    print(res.status_code)
    print(res.content)

def getTemplateDslJson():
    username = "elastic"
    password = "123456"
    user_info_str = username + ":" + password
    user_info = base64.b64encode(user_info_str.encode())  # 这个得到是个字节类型的数据
    headers = {
        "Authorization": "Basic {0}".format(user_info.decode())  # 这个就是需要验证的信息
    }
    url = "http://192.168.0.1:9200/_template/*_template"
    res = requests.get(url, headers=headers)
    print(res.status_code)
    return json.loads(res.content)

if __name__ == '__main__':
    jsonTemplate = getTemplateDslJson()
    if isinstance(jsonTemplate, dict):
        for templateName in jsonTemplate:
            templateDslJson = jsonTemplate[templateName]
            putTemplate(templateName, templateDslJson)