ElasticSearch

发表于 | 更新于: | 分类于 | | 本文总阅读量 次

一. Lucene

查询总结:

  • Term 查询(传入new Term()对象), 不分词, 严格匹配内容
  • 分词查询
  • 多字段查询
  • Boolean 查询
  • 范围查询 (数字类型)
  • 确切值查询 (数字类型)
  • 查询所有

索引 和 Document 都存在磁盘

代码见我的 github

二. Nginx 整合KeepAlive

image-20180813153414201

三. ECharts

散点图, 堆叠区域图, 饼图, 热力图

  1. 登录官网 http://echarts.baidu.com/

  2. 查看实例->官方实例->随便看看

  3. 查看 文档-> 教程 http://echarts.baidu.com/tutorial.html#5%20%E5%88%86%E9%92%9F%E4%B8%8A%E6%89%8B%20ECharts

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    1. 引入 Echarts, 创建 html 文件 hello.html
        <!DOCTYPE html>
        <html>
        <head>
            <meta charset="utf-8">
            <!-- 引入 ECharts 文件 -->
            <script src="echarts.min.js"></script>
        </head>
        </html>

    2. 下载 echarts.min.js 或其他版本 http://echarts.baidu.com/download.html,  放在 hello.html 的同级目录下
    3. 创建容器
        <body>
            <!-- 为 ECharts 准备一个具备大小（宽高）的 DOM -->
            <div id="main" style="width: 600px;height:400px;"></div>
        </body>

    4. 通过 echarts.init 方法初始化一个 echarts 实例并通过 setOption 方法, 生成任何想要生成的内容
        <script type="text/javascript">
            // 基于准备好的dom，初始化echarts实例
            var myChart = echarts.init(document.getElementById('main'));

            // 指定图表的配置项和数据
            var option = {
               ......
            };

            // 使用刚指定的配置项和数据显示图表。
            myChart.setOption(option);
        </script>

四. 热力图

参考百度地图官方热力图案例

  • 可以直接使用百度开放平台的热力图, 申请 ak(秘钥) (accesskey)
  • 把 ak 放到百度的 js 的后面的参数
  • 加入一个热力图的 js
  • 其它的参考官方案例慢慢调

五. ElasticSearch

版本: 5.4.3, june27,2017

CentOS7 命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#查看服务状态
systemctl status NetworkManager / network
#停止服务
systemctl stop NetworkManager
#启动服务
systemctl start NetworkManager
#禁止服务开机启动
systemctl disable NetworkManager
#设置服务开机启动
systemctl enable NetworkManager

#查看网络信息
ip addr
#查看具体某个网卡信息
ip addr ls eno16777736

#停止网卡
ip link set eno16777736 down
#启动网卡
ip link set eno16777736 up

#修改ip地址
ip addr del 192.168.80.134/24 dev eno16777736
ip addr add 192.168.80.136/24 dev eno16777736

#查看路由信息
ip route show
#添加路由
ip route add default via 192.168.80.2 dev br0

#安装ifconfig命令
yum install -y net-tools


#修改主机名
hostnamectl set-hostname node-1




#防火墙配置:Centos升级到7之后，发现无法使用iptables控制Linux的端口，google之后发现Centos 7使用firewalld代替了原来的iptables。下面记录如何使用firewalld开放Linux端口： 

#开启防火墙
systemctl start firewalld

#开启端口
firewall-cmd --zone=public --add-port=08/tcp --permanent
 
命令含义：
--zone #作用域
--add-port=80/tcp  #添加端口，格式为：端口/通讯协议
--permanent   #永久生效，没有此参数重启后失效

#重新加载防火墙配置
firewall-cmd --reload

安装单机版

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
0. 使用 centos7
1. 安装JDK（1.8）
2. 上传解压Elasticsearch-5.4.3
3. 创建一个普通用户，然后将对于的目录修改为普通用户的所属用户和所属组
4. 修改配置文件config/elasticsearch.yml 的 network.host 为安装es 主机的 host
	network.host: 192.168.170.171	#修改	
	transport.tcp.port: 9300    	#添加
	discovery.zen.ping.unicast.hosts: ["192.168.170.171:9300"]		#修改
	
5. 启动ES，发现报错
	bin/elasticsearch
	#出现错误
	[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]
	[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

	#用户最大可创建文件数太小
	sudo vi /etc/security/limits.conf
	* soft nofile 65536
	* hard nofile 65536

	#查看可打开文件数量
	ulimit -Hn

	#最大虚拟内存太小
	sudo vi /etc/sysctl.conf 
	vm.max_map_count=262144

	#查看虚拟内存的大小
	sudo sysctl -p

6. 重启linux
	shutdown -r now
	关机
    shutdown -h now

7. 通过浏览器访问ES
	192.168.170.171:9200

安装集群

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
1.安装jdk（jdk要求1.8.20以上）

2.上传es安装包

3.解压es
tar -zxvf elasticsearch-5.4.3.tar.gz -C apps/

4.修改配置
vi aps/es/config/elasticsearch.yml

#集群名称，通过组播的方式通信，通过名称判断属于哪个集群
cluster.name: my-es

#节点名称，要唯一
node.name: node-1

#数据存放位置
path.data: /home/ap/data/es/data

#日志存放位置(可选)
path.logs: /home/ap/data/es/logs

#es绑定的ip地址
network.host: cs7_1

#es默认端口
http.port: 9200
transport.tcp.port: 9300

#初始化时可进行选举的节点
discovery.zen.ping.unicast.hosts: ["cs7_1:9300","cs7_2:9300","cs7_3:9300"]

#后台启动
$~> elasticsearch/bin/elasticsearch -d

5.使用scp拷贝到其他节点
scp elasticsearch.yml cs7_2:$PWD
scp elasticsearch.yml cs7_3:$PWD

6.在其他节点上修改es配置，需要修改的有node.name和network.host

7.启动es（elasticsearch/bin/elasticsearch -h查看帮助文档） 
$~> apps/es/bin/elasticsearch -d


8.用浏览器访问es所在机器的9200端口
http://192.168.170.171:9200/ 或者 http://cs7_1:9200 
{
  "name" : "node-1",
  "cluster_name" : "my-es",
  "cluster_uuid" : "9jgURtc7TOejQnumbZt7VA",
  "version" : {
    "number" : "5.4.3",
    "build_hash" : "eed30a8",
    "build_date" : "2017-06-22T00:34:03.743Z",
    "build_snapshot" : false,
    "lucene_version" : "6.5.1"
  },
  "tagline" : "You Know, for Search"
}

# 杀死 ES 进程, 关闭进程
# 后面是查出进程号, 相当于 kill xxx
kill `ps -ef | grep Elasticsearch | grep -v grep | awk '{print $2}'`

#查看集群状态
curl -XGET 'http://cs7_1:9200/_cluster/health?pretty'
{
  "cluster_name" : "my-es",
  "status" : "green",
  "timed_out" : false,
  "number_of_nodes" : 3,
  "number_of_data_nodes" : 3,
  "active_primary_shards" : 0,
  "active_shards" : 0,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 0,
  "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" : 100.0
}
# 浏览器中查看集群状态 - json
http://cs7_1:9200/_cluster/health?pretty




--------------------------------------------
###注意: 此错误已在上面安装单机的过程中解决掉
#出现错误 
[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]
[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

#用户最大可创建文件数太小
sudo vi /etc/security/limits.conf
* soft nofile 65536
* hard nofile 65536

#查看可打开文件数量(重启后查询)
ulimit -Hn


#最大虚拟内存太小
sudo vi /etc/sysctl.conf 
vm.max_map_count=262144

#查看虚拟内存的大小
sudo sysctl -p
------------

ES 插件 ElasticSearch-Header 安装

9100端口

如果访问不了, 可以先查看端口, 通过端口查到进程, kill 掉, 再重启 

1
2
3
yum install -y lsof
lsof -i tcp:9100
# 找到 pid 后, kill 掉, 再重新启动

参考博客

image-20180815093732538

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
0> yum 源首先改为国内的


#更新
sudo yum update -y

#安装扩展源
sudo rpm -ivh http://dl.fedoraproject.org/pub/epel/epel-release-latest-7.noarch.rpm
sudo rpm -ivh https://kojipkgs.fedoraproject.org//packages/http-parser/2.7.1/3.el7/x86_64/http-parser-2.7.1-3.el7.x86_64.rpm

#安装 js 的包管理工具 npm
sudo yum install-y npm

#安装 git
sudo yum install -y git

#安装 bzip2
sudo yum install -y bzip2

#下载 elasticsearch-head
git clone git://github.com/mobz/elasticsearch-head.git

#将源码包下载后剪切到/bigdata目录，并改所属用户和组
sudo chown -R ap:ap /home/ap/apps/elasticsearch-head

#进入到elasticsearch-head中
cd elasticsearch-head

#编译安装
npm install


打开elasticsearch-head-master/Gruntfile.js，找到下面connect属性，新增hostname: '0.0.0.0',
		connect: {
                        server: {
                                options: {
                                        hostname: '0.0.0.0',
                                        port: 9100,
                                        base: '.',
                                        keepalive: true
                                }
                        }
                }

编辑elasticsearch-5.4.3/config/elasticsearch.yml,加入以下内容：
http.cors.enabled: true
http.cors.allow-origin: "*"

#运行服务
npm run start

安装 IK 分词器

github 地址 https://github.com/medcl/elasticsearch-analysis-ik/releases

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
# 安装 & 测试
安装IK分词器
下载对应版本的插件
https://github.com/medcl/elasticsearch-analysis-ik/releases


首先下载es对应版本的ik分词器的zip包，上传到es服务器上，在es的安装目录下有一个plugins的目录，在这个目录下创建一个叫ik的目录
然后将解压好的内容，拷贝到ik目录
将ik目录拷贝到其他的es节点
重新启动所有的es


#创建索引名字叫news
curl -XPUT http://192.168.170.171:9200/news

#创建mapping（相当于数据中的schema信息，表名和字段名以及字段的类型）
curl -XPOST http://192.168.170.171:9200/news/fulltext/_mapping -d'
{
        "properties": {
            "content": {
                "type": "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_max_word"
            }
        }
    
}'


curl -XPOST http://192.168.170.171:9200/news/fulltext/1 -d'
{"content":"美国留给伊拉克的是个烂摊子吗"}'

curl -XPOST http://192.168.170.171:9200/news/fulltext/2 -d'
{"content":"公安部：各地校车将享最高路权"}'

curl -XPOST http://192.168.170.171:9200/news/fulltext/3 -d'
{"content":"中韩渔警冲突调查：韩警平均每天扣1艘中国渔船"}'

curl -XPOST http://192.168.170.171:9200/news/fulltext/4 -d'
{"content":"中国驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"}'

curl -XPOST http://192.168.170.171:9200/news/fulltext/_search  -d'
{
    "query" : { "match" : { "content" : "中国" }},
    "highlight" : {
        "pre_tags" : ["<font color='red'>", "<tag2>"],
        "post_tags" : ["</font>", "</tag2>"],
        "fields" : {
            "content" : {}
        }
    }
}'

-------------------------------------------------------------------


curl -XGET 'http://192.168.170.171:9200/_analyze?pretty&analyzer=ik_max_word' -d '联想是全球最大的笔记本厂商'

# 分的更准, 建议使用
curl -XGET 'http://192.168.170.171:9200/_analyze?pretty&analyzer=ik_smart' -d '联想是全球最大的笔记本厂商'

curl -XPUT 'https://192.168.170.171:9200/iktest?pretty' -d '{
    "settings" : {
        "analysis" : {
            "analyzer" : {
                "ik" : {
                    "tokenizer" : "ik_max_word"
                }
            }
        }
    },
    "mappings" : {
        "article" : {
            # dynamic 为 true, 可以指定多个字段
            "dynamic" : true,      
            "properties" : {
                "subject" : {
                    "type" : "string",
                    "analyzer" : "ik_max_word"
                }
            }
        }
    }
}'

curl -XPUT 'https://192.168.170.171:9200/iktest?pretty' -d '{
    "settings" : {
        "analysis" : {
            "analyzer" : {
                "ik" : {
                    "tokenizer" : "ik_max_word"
                }
            }
        }
    },
    "mappings" : {
        "article" : {
            "dynamic" : true,
            "properties" : {
                "subject" : {
                    "type" : "string",
                    "analyzer" : "ik_max_word"
                }
            }
        }
    }
}'


curl -XGET 'http://192.168.170.171:9200/_analyze?pretty&analyzer=ik_max_word' -d ‘中华人民共和国’

ES 的 JavaAPI 和 聚合查询

可以参考我的 github

1. helloworld
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package com.rox.es;

import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;

public class HelloWorldES {

    public static void main(String[] args) {

        try {

            //设置集群名称
            Settings settings = Settings.builder()
                    .put("cluster.name", "my-es")
                    .build();
            //创建client
            TransportClient client = new PreBuiltTransportClient(settings).addTransportAddresses(
                    //用java访问ES用的端口是9300
                    new InetSocketTransportAddress(InetAddress.getByName("192.168.170.171"), 9300),
                    new InetSocketTransportAddress(InetAddress.getByName("192.168.170.172"), 9300),
                    new InetSocketTransportAddress(InetAddress.getByName("192.168.170.173"), 9300)
            );
            //搜索数据（.actionGet()方法是同步的，没有返回就等待）
            GetResponse response = client.prepareGet("news", "fulltext", "1").execute().actionGet();
            //输出结果
            System.out.println(response);
            //关闭client
            client.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
2. ES 的 CRUD 增删改查
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
package com.rox.es;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.byscroll.BulkByScrollResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.DoubleTerms;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
import org.elasticsearch.search.aggregations.metrics.max.InternalMax;
import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.InetAddress;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;


/**
 * https://www.elastic.co/guide/en/elasticsearch/client/java-api/5.4/index.html
 */
public class EsCRUD {

    private TransportClient client = null;

    @Before
    public void init() throws Exception {
        //设置集群名称
        Settings settings = Settings.builder()
                .put("cluster.name", "my-es")
                //自动感知的功能（可以通过当前指定的节点获取所有es节点的信息）
                .put("client.transport.sniff", true)
                .build();
        //创建client
        client = new PreBuiltTransportClient(settings).addTransportAddresses(
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.170"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.171"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.172"), 9300));

    }


    /**
     * 新增
     * @throws IOException
     */
    @Test
    public void testCreate() throws IOException {

        IndexResponse response = client.prepareIndex("gamelog", "users", "1")
                .setSource(
                        jsonBuilder()
                                .startObject()
                                    .field("username", "老李")
                                    .field("gender", "male")
                                    .field("birthday", new Date())
                                    .field("fv", 999)
                                    .field("message", "老李还行")
                                .endObject()
                ).get();


    }

    //查找一条
    @Test
    public void testGet() throws IOException {
        GetResponse response = client.prepareGet("gamelog", "users", "1").get();
        System.out.println(response.getSourceAsString());
    }


    //查找多条
    @Test
    public void testMultiGet() throws IOException {
        MultiGetResponse multiGetItemResponses = client.prepareMultiGet()
                .add("gamelog", "users", "1")
                .add("gamelog", "users", "2", "3")
                .add("news", "fulltext", "1")
                .get();

        for (MultiGetItemResponse itemResponse : multiGetItemResponses) {
            GetResponse response = itemResponse.getResponse();
            if (response.isExists()) {
                String json = response.getSourceAsString();
                System.out.println(json);
            }
        }
    }

    /**
     * 更新 (改)
     * @throws Exception
     */
    @Test
    public void testUpdate() throws Exception {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index("gamelog");
        updateRequest.type("users");
        updateRequest.id("2");
        updateRequest.doc(
                jsonBuilder()
                        .startObject()
                        .field("fv", 999.9)
                        .endObject());
        client.update(updateRequest).get();
    }

    /**
     * 删除
     */
    @Test
    public void testDelete() {
        DeleteResponse response = client.prepareDelete("gamelog", "users", "1").get();
        System.out.println(response);
    }


    /**
     * 查询出结果 -> 删除
     */
    @Test
    public void testDeleteByQuery() {
        BulkByScrollResponse response =
                DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
                        //指定查询条件
                        .filter(QueryBuilders.matchQuery("username", "老段"))
                        //指定索引名称
                        .source("gamelog")
                        .get();
        long deleted = response.getDeleted();
        System.out.println(deleted);
    }


    //异步删除
    @Test
    public void testDeleteByQueryAsync() {
        DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
                .filter(QueryBuilders.matchQuery("gender", "male"))
                .source("gamelog")
                .execute(new ActionListener<BulkByScrollResponse>() {
                    @Override
                    public void onResponse(BulkByScrollResponse response) {
                        long deleted = response.getDeleted();
                        System.out.println("数据删除了");
                        System.out.println(deleted);
                    }

                    @Override
                    public void onFailure(Exception e) {
                        e.printStackTrace();
                    }
                });
        try {
            System.out.println("异步删除");
            Thread.sleep(10000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 范围查询
     * 注意: 这里如果前面设置的是 doule 类型, 搜索的范围也是 double 类型
     */
    @Test
    public void testRange() {

        QueryBuilder qb = rangeQuery("fv")
                // [88.99, 10000)
                .from(88.99)
                .to(100000)
                .includeLower(true)
                .includeUpper(false);

        SearchResponse response = client.prepareSearch("gamelog").setQuery(qb).get();

        System.out.println(response);
    }


    /**
     * NBA相关查询
     * curl -XPUT 'http://192.168.5.251:9200/player_info/player/1' -d '{ "name": "curry", "age": 29, "salary": 3500,"team": "war", "position": "pg"}'
     * curl -XPUT 'http://192.168.5.251:9200/player_info/player/2' -d '{ "name": "thompson", "age": 26, "salary": 2000,"team": "war", "position": "pg"}'
     * curl -XPUT 'http://192.168.5.251:9200/player_info/player/3' -d '{ "name": "irving", "age": 25, "salary": 2000,"team": "cav", "position": "pg"}'
     * curl -XPUT 'http://192.168.5.251:9200/player_info/player/4' -d '{ "name": "green", "age": 26, "salary": 2000,"team": "war", "position": "pf"}'
     * curl -XPUT 'http://192.168.5.251:9200/player_info/player/5' -d '{ "name": "james", "age": 33, "salary": 4000,"team": "cav", "position": "sf"}'
     */

    // 创建 index 和 type
    @Test
    public void testAddPlayer() throws IOException {

        IndexResponse response = client.prepareIndex("player_info", "player", "1")
                .setSource(
                        jsonBuilder()
                                .startObject()
                                .field("name", "James")
                                .field("age", 33)
                                .field("salary", 3000)
                                .field("team", "cav")
                                .field("position", "sf")
                                .endObject()
                ).get();
    }

    /**
     * 复杂条件查询
     * https://elasticsearch.cn/article/102
     *
     * select team, count(*) as player_count from player group by team;
     */
    @Test
    public void testAgg1() {

        //指定索引和type
        SearchRequestBuilder builder = client.prepareSearch("player_info").setTypes("player");
        //按team分组然后聚合，但是并没有指定聚合函数
        TermsAggregationBuilder teamAgg = AggregationBuilders.terms("player_count").field("team");
        //添加聚合器
        builder.addAggregation(teamAgg);
        //触发
        SearchResponse response = builder.execute().actionGet();
        //System.out.println(response);
        //将返回的结果放入到一个map中
        Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();
//        Set<String> keys = aggMap.keySet();
//
//        for (String key: keys) {
//            System.out.println(key);
//        }

//        //取出聚合属性
        StringTerms terms = (StringTerms) aggMap.get("player_count");


        //
////        //依次迭代出分组聚合数据
//        for (Terms.Bucket bucket : terms.getBuckets()) {
//            //分组的名字
//            String team = (String) bucket.getKey();
//            //count，分组后一个组有多少数据
//            long count = bucket.getDocCount();
//            System.out.println(team + " " + count);
//        }

        Iterator<Terms.Bucket> teamBucketIt = terms.getBuckets().iterator();
        while (teamBucketIt .hasNext()) {
            Terms.Bucket bucket = teamBucketIt.next();
            String team = (String) bucket.getKey();

            long count = bucket.getDocCount();

            System.out.println(team + " " + count);
        }

    }

    /**
     * select team, position, count(*) as pos_count from player group by team, position;
     * 先按 team, team 中再按 position 聚合
     */
    @Test
    public void testAgg2() {
        SearchRequestBuilder builder = client.prepareSearch("player_info").setTypes("player");
        //指定别名和分组的字段
        TermsAggregationBuilder teamAgg = AggregationBuilders.terms("team_name").field("team");
        TermsAggregationBuilder posAgg= AggregationBuilders.terms("pos_count").field("position");
        //添加两个聚合构建器
        /**
         * 注意: 此处是在 teamAgg 上, 添加 subAggregation(posAgg)
         */
        builder.addAggregation(teamAgg.subAggregation(posAgg));
        //执行查询
        SearchResponse response = builder.execute().actionGet();
        //将查询结果放入map中
        Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();
        //根据属性名到map中查找
        StringTerms teams = (StringTerms) aggMap.get("team_name");
        //循环查找结果
        for (Terms.Bucket teamBucket : teams.getBuckets()) {
            //先按球队进行分组
            String team = (String) teamBucket.getKey();
            Map<String, Aggregation> subAggMap = teamBucket.getAggregations().getAsMap();
            StringTerms positions = (StringTerms) subAggMap.get("pos_count");
            //因为一个球队有很多位置，那么还要依次拿出位置信息
            for (Terms.Bucket posBucket : positions.getBuckets()) {
                //拿到位置的名字
                String pos = (String) posBucket.getKey();
                //拿出该位置的数量
                long docCount = posBucket.getDocCount();
                //打印球队，位置，人数
                System.out.println(team + " " + pos + " " + docCount);
            }
        }

        /**
         * 结果:
         war pg 2
         war pf 1
         cav pg 1
         cav sf 1
         */
    }


    /**
     * select team, max(age) as max_age from player group by team;
     * 根据 team 名聚合后, 求聚合成员的 max age
     */
    @Test
    public void testAgg3() {

        // 指定 索引&类型  index & type
        SearchRequestBuilder builder = client.prepareSearch("player_info").setTypes("player");
        /**
         * term: 根据指定的名字创建一个新的聚合体
         * 指定按球队进行分组
         */
        TermsAggregationBuilder teamAgg = AggregationBuilders.terms("team_name").field("team");

        //指定分组求最大值, 指定聚合函数(max)和需要聚合的字段(age)
        MaxAggregationBuilder maxAgg = AggregationBuilders.max("max_age").field("age");

        //分组后求最大值(添加聚合器--聚合规则)
        builder.addAggregation(teamAgg.subAggregation(maxAgg));

        //查询
        SearchResponse response = builder.execute().actionGet();

        // 以 map 的形式获取聚合结果
        Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();

        //根据team属性为 key，获取map中的内容-value
        StringTerms teams = (StringTerms) aggMap.get("team_name");

        for (Terms.Bucket teamBucket : teams.getBuckets()) {

            //分组的属性名--有很多对, 把队名取出来
            String team = (String) teamBucket.getKey();

            //在将聚合后取最大值的内容取出来放到map中
            Map<String, Aggregation> subAggMap = teamBucket.getAggregations().getAsMap();

            //取分组后的最大值
            InternalMax ages = (InternalMax)subAggMap.get("max_age");
            double max = ages.getValue();
            System.out.println(team + " " + max);
        }
    }


    /**
     * select team, avg(age) as avg_age, sum(salary) as total_salary from player group by team;
     * team分组 下求 avg & sum
     */
    @Test
    public void testAgg4() {
        SearchRequestBuilder builder = client.prepareSearch("player_info").setTypes("player");
        //指定分组字段
        TermsAggregationBuilder termsAgg = AggregationBuilders.terms("team_name").field("team");
        //指定聚合函数是求平均数据
        AvgAggregationBuilder avgAgg = AggregationBuilders.avg("avg_age").field("age");
        //指定另外一个聚合函数是求和
        SumAggregationBuilder sumAgg = AggregationBuilders.sum("total_salary").field("salary");

        //分组的聚合器关联了两个聚合函数(team 下关联了 avg & sum)
        builder.addAggregation(termsAgg.subAggregation(avgAgg).subAggregation(sumAgg));

        SearchResponse response = builder.execute().actionGet();

        Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();

        //按分组的名字取出数据
        StringTerms teams = (StringTerms) aggMap.get("team_name");
        for (Terms.Bucket teamBucket : teams.getBuckets()) {
            //获取球队名字
            String team = (String) teamBucket.getKey();
            Map<String, Aggregation> subAggMap = teamBucket.getAggregations().getAsMap();
            //根据别名取出平均年龄
            InternalAvg avgAge = (InternalAvg)subAggMap.get("avg_age");
            //根据别名取出薪水总和
            InternalSum totalSalary = (InternalSum)subAggMap.get("total_salary");
            double avgAgeValue = avgAge.getValue();
            double totalSalaryValue = totalSalary.getValue();
            System.out.println(team + " " + avgAgeValue + " " + totalSalaryValue);
        }
    }


    /**
     * select team, sum(salary) as total_salary from player group by team order by total_salary desc;
     * 按 team 聚合, 按sum salary 降序排列
     */
    @Test
    public void testAgg5() {
        SearchRequestBuilder builder = client.prepareSearch("player_info").setTypes("player");
        //按team进行分组，然后指定排序规则  asc 升序, desc 降序
        //排序规则在分组时就指定了
        TermsAggregationBuilder termsAgg = AggregationBuilders.terms("team_name").field("team").order(Terms.Order.aggregation("total_salary ", false));
        SumAggregationBuilder sumAgg = AggregationBuilders.sum("total_salary").field("salary");
        builder.addAggregation(termsAgg.subAggregation(sumAgg));
        SearchResponse response = builder.execute().actionGet();
        Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();
        StringTerms teams = (StringTerms) aggMap.get("team_name");
        for (Terms.Bucket teamBucket : teams.getBuckets()) {
            String team = (String) teamBucket.getKey();
            Map<String, Aggregation> subAggMap = teamBucket.getAggregations().getAsMap();
            InternalSum totalSalary = (InternalSum)subAggMap.get("total_salary");
            double totalSalaryValue = totalSalary.getValue();
            System.out.println(team + " " + totalSalaryValue);
        }
    }
}
3. ES 的 Admin API
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
package com.rox.es;

import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.InetAddress;
import java.util.HashMap;

/**
 *
 */
public class AdminAPI {

    private TransportClient client = null;

    //在所有的测试方法之前执行
    @Before
    public void init() throws Exception {
        //设置集群名称
        Settings settings = Settings.builder().put("cluster.name", "my-es").build();
        //创建client
        client = new PreBuiltTransportClient(settings).addTransportAddresses(
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.171"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.172"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.170.173"), 9300));

    }

    //创建索引，并配置一些参数
    //创建 再由索引创建 type
    @Test
    public void createIndexWithSettings() {
        //获取Admin的API
        AdminClient admin = client.admin();
        //使用Admin API对索引进行操作 拿到目录
        IndicesAdminClient indices = admin.indices();
        //准备创建索引
        indices.prepareCreate("gamelog")
                //配置索引参数
                .setSettings(
                        //参数配置器
                        Settings.builder()//指定索引分区的数量
                                .put("index.number_of_shards", 4)
                                //指定索引副本的数量（注意：不包括本身，如果设置数据存储副本为2，实际上数据存储了3份）
                                .put("index.number_of_replicas", 2)
                )
                //真正执行
                .get();
    }

    //跟索引添加mapping信息（给表添加schema信息）
    @Test
    public void putMapping() {
        //创建索引
        client.admin().indices().prepareCreate("twitter")
                //创建一个type，并指定type中属性的名字和类型
                .addMapping("tweet",
                        "{\n" +
                                "    \"tweet\": {\n" +
                                "      \"properties\": {\n" +
                                "        \"message\": {\n" +
                                "          \"type\": \"string\"\n" +
                                "        }\n" +
                                "      }\n" +
                                "    }\n" +
                                "  }")
                .get();
    }

    /**
     * 生产中用的!!
     * 你可以通过dynamic设置来控制这一行为，它能够接受以下的选项： ※
     * true：默认值。动态添加字段
     * false：忽略新字段
     * strict：如果碰到陌生字段，抛出异常
     *
     * @throws IOException
     */
    @Test
    public void testSettingsMappings() throws IOException {
        //1:settings
        HashMap<String, Object> settings_map = new HashMap<String, Object>(2);
        settings_map.put("number_of_shards", 3);
        settings_map.put("number_of_replicas", 2);

        //2:mappings（映射、schema）
        XContentBuilder builder = XContentFactory.jsonBuilder()
                .startObject()
                    .field("dynamic", "true")
                    //设置type中的属性
                    .startObject("properties")
                        //num属性
                        .startObject("num")
                            //类型是integer
                            .field("type", "integer")
                            //不分词，但是建索引
                            .field("index", "not_analyzed")
                            //在文档中存储
                            .field("store", "yes")
                        .endObject()
                        //name属性
                        .startObject("name")
                            //string类型
                            .field("type", "string")
                            //在文档中存储
                            .field("store", "yes")
                            //分词, 建立索引
                            .field("index", "analyzed")
                            //使用ik_smart进行分词
                            .field("analyzer", "ik_smart")
                        .endObject()
                    .endObject()
                .endObject();

        CreateIndexRequestBuilder prepareCreate = client.admin().indices().prepareCreate("user_info");
        //管理索引（user_info）然后关联type（user）
        prepareCreate.setSettings(settings_map).addMapping("user", builder).get();
    }

    /**
     * XContentBuilder mapping = jsonBuilder()
     .startObject()
     .startObject("productIndex")
     .startObject("properties")
     .startObject("title").field("type", "string").field("store", "yes").endObject()
     .startObject("description").field("type", "string").field("index", "not_analyzed").endObject()
     .startObject("price").field("type", "double").endObject()
     .startObject("onSale").field("type", "boolean").endObject()
     .startObject("type").field("type", "integer").endObject()
     .startObject("createDate").field("type", "date").endObject()
     .endObject()
     .endObject()
     .endObject();
     PutMappingRequest mappingRequest = Requests.putMappingRequest("productIndex").type("productIndex").source(mapping);
     client.admin().indices().putMapping(mappingRequest).actionGet();
     */


    /**
     * index这个属性，no代表不建索引
     * not_analyzed，建索引不分词
     * analyzed 即分词，又建立索引
     * expected [no], [not_analyzed] or [analyzed]
     *
     * @throws IOException
     */
    // 实战案例写法
    @Test
    public void testSettingsPlayerMappings() throws IOException {
        //1:settings
        HashMap<String, Object> settings_map = new HashMap<String, Object>(2);
        settings_map.put("number_of_shards", 3);
        settings_map.put("number_of_replicas", 1);

        //2:mappings
        XContentBuilder builder = XContentFactory.jsonBuilder()
                .startObject()//
                        .field("dynamic", "true")
                    .startObject("properties")
                        .startObject("id")
                            .field("type", "integer")
                            .field("store", "yes")
                        .endObject()

                        .startObject("name")
                            .field("type", "string")
                            // 只要不是 index, no,  就是建索引
                            .field("index", "not_analyzed")
                        .endObject()

                        .startObject("age")
                            .field("type", "integer")
                        .endObject()

                        .startObject("salary")
                            .field("type", "integer")
                        .endObject()

                        .startObject("team")
                            .field("type", "string")
                            .field("index", "not_analyzed")
                        .endObject()

                        .startObject("position")
                            .field("type", "string")
                            .field("index", "not_analyzed")
                        .endObject()

                        .startObject("description")
                            .field("type", "string")
                            .field("store", "no")
                            .field("index", "analyzed")
                            .field("analyzer", "ik_smart")
                        .endObject()

                        .startObject("addr")
                            .field("type", "string")
                            .field("store", "yes")
                            .field("index", "analyzed")
                            .field("analyzer", "ik_smart")
                        .endObject()
                    .endObject()
                .endObject();

        CreateIndexRequestBuilder prepareCreate = client.admin().indices().prepareCreate("player_info");
        prepareCreate.setSettings(settings_map).addMapping("player", builder).get();

    }
}

ES 安装 SQL 插件 & 图形化界面插件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#es安装SQL插件
./bin/elasticsearch-plugin install https://github.com/NLPchina/elasticsearch-sql/releases/download/5.4.3.0/elasticsearch-sql-5.4.3.0.zip

#然后将解压到plugins目录下的内容拷贝到其他es的节点的plugins目录, 重启 es, 就可以使用 sql 风格查询了, 类似于下面, 直接在浏览器输入:
http://cs7_1:9200/_sql?sql=select * from player_info limit 10
http://cs7_1:9200/_sql?sql=select team, avg(salary) avg_sal from player_info group by team


#下载SQL的Server 图形化界面插件
wget https://github.com/NLPchina/elasticsearch-sql/releases/download/5.4.1.0/es-sql-site-standalone.zip

#用npm编译安装
unzip es-sql-site-standalone.zip
cd site-server/
npm install express --save

#修改SQL的Server的端口, 默认是8080
vi site_configuration.json
启动服务
node node-server.js &

>> 注意: 图形化界面工具, 右上角的框中, 要填上连接 es 服务器的地址. 
 		此外, 还有 explain 功能, 可以把 sql 转化为restful 风格的 json, 如下图所示

image-20180815000912709

如果帮到你, 可以给我赞助杯咖啡☕️
0%