elasticsearch环境搭建与使用

1.ES的安装与环境配置

首先请先去了解下ES的一些基本概念：cluster，node，index，shard，replica shard，plugin，river；这里就不一一赘述了。

ES的安装很简单：

    –  确保机器上已经安装了JDK7以上版本

    –  下载：官网下载地址：https://www.elastic.co/downloads/elasticsearch

    –  将下载后的文件解压到/opt/eshome(默认的es home，可更改)

    –  进入到bin文件夹下，执行“./elasticsearch -d”(-d表示在后台执行)   

安装结束后，访问  http://ip:9200/，出现下面的返回结果。默认的cluster名为elasticsearch

{

  "status" : 200,

  "name" : "test_node_196",

  "cluster_name" : "test_cluster1",

  "version" : {

    "number" : "1.7.0",

    "build_hash" : "929b9739cae115e73c346cb5f9a6f24ba735a743",

    "build_timestamp" : "2015-07-16T14:31:07Z",

    "build_snapshot" : false,

    "lucene_version" : "4.10.4"

  },

  "tagline" : "You Know, for Search"

} 

 

ES环境配置

ES的配置文件在conf/elasticsearch.yml。

master节点、data节点和river节点都在该配置文件里面配备。（river节点在后面介绍）

 

master节点的配备如下：

cluster.name: test_cluster1    ##cluster名

node.name: "test_node_196"    ##节点名称

node.master: true   ##是否是master节点

node.data: true   ##该节点上是否保存数据

index.number_of_replicas: 1   ##备份的数量，这里设为1

path.data: /opt/esdata    ##该节点上数据存储的path

transport.tcp.port: 9300   ##tcp的端口号

http.port: 9200    ##http的端口号

discovery.zen.minimum_master_nodes: 1

discovery.zen.ping.timeout: 3s    ##节点间自动发现的响应时间

discovery.zen.ping.unicast.hosts: ["localhost"]    ##节点间自动发现，master节点为localhost

 

data节点的配备如下

cluster.name: test_cluster1    ##cluster名，注意和master节点保持一致

node.name: "test_node_197"

node.master: false   ##不是master节点

node.data: true

index.number_of_replicas: 1

path.data: /opt/esdata

transport.tcp.port: 9300

http.port: 9200

discovery.zen.minimum_master_nodes: 1

discovery.zen.ping.timeout: 3s    ##节点间自动发现的响应时间

discovery.zen.ping.unicast.hosts: ["10.19.220.196"]    ##节点间自动发现，这里要写master节点的IP，多个要用逗号隔开

注意：1.为了测试，多个node可以在同一台服务器上启动，但通常一个服务器只放一个node。

           2.系统启动时，node会使用广播来发现一个现有的cluster，并且试图加入该cluster

           3.配置文件中的冒号后要加空格，否则启动es时会报load配置文件发生错误

           4.为了防止脑裂，需要将discovery.zen.minimum_master_nodes设为[master节点数/2+1]。

              关于脑裂，请参见http://blog.trifork.com/2013/10/24/how-to-avoid-the-split-brain-problem-in-elasticsearch/

           5.当将一个节点的index.number_of_replicas从0变为1时，ES中现有的索引的分片仍然是没有备份的，但修改后新建的索引的分片都有一个备份

 

使用curl操作es

下面是一些通过curl操作es的例子，_cat用于查看，_update用于更新，_bulk用于批量操作，_search用于查询。

具体的就不一一解释了。可参见官网：https://www.elastic.co/guide/en/elasticsearch/reference/current/getting-started.html

##To check the cluster health

curl 'localhost:9200/_cat/health?v'

 

##get a list of nodes in our cluster

curl 'localhost:9200/_cat/nodes?v'

 

##take a peek at our indices

curl 'localhost:9200/_cat/indices?v'

 

##create an index named "customer"

curl -XPUT 'localhost:9200/customer?pretty'

 

##index a simple customer document into the customer index, "external" type, with an ID of 1

curl -XPUT 'localhost:9200/customer/external/1?pretty'-d ' { "name": "John Doe" }'

 

##retrieve that document

curl -XGET 'localhost:9200/customer/external/1?pretty'

 

##delete the index

curl -XDELETE 'localhost:9200/bankfortest?pretty'

 

##update our previous document

curl -XPOST 'localhost:9200/customer/external/1/_update?pretty'-d ' { "doc": { "name": "Jane Doe", "age": 20 } }'

##注意：要运行下面的语句，需要在配置文件中加上“script.disable_dynamic: false”

curl -XPOST 'localhost:9200/customer/external/1/_update?pretty'-d ' { "script" : "ctx._source.age += 5" }'

 

##delete multiple documents that match a query condition

curl -XDELETE 'localhost:9200/customer/external/_query?pretty'-d ' { "query": { "match": { "name": "John" } } }'

 

##批量操作

curl -XPOST 'localhost:9200/customer/external/_bulk?pretty'-d ' {"index":{"_id":"5"}} {"name": "John Doe" } {"index":{"_id":"6"}} {"name": "Jane Doe" } '

curl -XPOST 'localhost:9200/customer/external/_bulk?pretty'-d ' {"update":{"_id":"1"}} {"doc": { "name": "John Doe becomes Jane Doe" } } {"delete":{"_id":"2"}} '

curl -XPOST 'localhost:9200/bankfortest/account/_bulk?pretty' --data-binary @accounts.json

 

##查询

curl 'localhost:9200/bank/_search?q=*&pretty'

curl -XPOST 'localhost:9200/bank/_search?pretty'-d ' { "query": { "match_all": {} } }'

curl -XPOST 'localhost:9200/bank/_search?pretty' -d '

{

  "query": { "match_all": {} },

  "_source": ["account_number", "balance"]

}'

curl -XPOST 'localhost:9200/bank/_search?pretty' -d '

{

  "query": { "match": { "account_number": 20 } }

}'

 

2.常用的插件head，bigdesk，essql

首先请到官网上下载这些插件，安装指令如下：

bin/plugin --url file:///path/to/plugin --install plugin-name

 

head插件相当于es的一个管理控制台，可以看到es中节点、数据等内容

访问地址：http://ip:9600/_plugin/head/

 

bigdesk插件相当于一个es的监控插件，访问地址http://ip:9200/_plugin/bigdesk/

 

essql插件是一个sql插件，你可以使用普通sql语句去访问es里面的数据。

访问地址：http://ip:9200/_plugin/essql/

 

 

3.从es中读取数据

es中数据读取有两种方法：一种是通过restful API，一种是通过TransportClient，这里主要介绍第二种方式，效率更高，更易编码。

 

下面是一些简单的增删改查的例子。

（具体的ES的Java API请参照官网https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-api.html）

public class EsIndexTest {

    public static Client client;

 

    @Before

    public void init() {

        TransportClient transportClient = new TransportClient();

        Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name", "test_cluster1").build();

        transportClient = new TransportClient(settings);

        //这里的端口号要与配置文件中的transport.tcp.port保持一致，否则会出现NoNodeAvailableException

        client = transportClient.addTransportAddress(new InetSocketTransportAddress(

                        "10.19.220.196", 9300));

    }

 

    public static DateTimeFormatter dateFormatter = DateTimeFormat.forPattern(

            "yyyy-MM-dd'T'HH:mm:ss.SSS").withZone(

            DateTimeZone.forOffsetHours(8));

 

    public static String obj2JsonData() {

        String jsonData = null;

        try {

            // 使用XContentBuilder创建json数据

            Date date = new Date();

            //Gson gson = new Gson();

            //System.err.println("date-->" + gson.toJson(date));

            XContentBuilder jsonBuilder = XContentFactory.jsonBuilder();

            jsonBuilder

                    .startObject()

                    .field("id", 78)

                    .field("name", "test")

                    .field("funciton", date, dateFormatter)

                    .endObject();

            jsonData = jsonBuilder.string();

            System.out.println(jsonData);

        } catch (IOException e) {

            e.printStackTrace();

        }

        return jsonData;

    }

 

    @Test

    //创建一个index，并向里面写入json数据

    public void createIndexResponse() {

        String jsonData = obj2JsonData();

        IndexRequestBuilder ib = client.prepareIndex("bank2", "account2").setSource(

                jsonData);

        IndexResponse response = ib.execute().actionGet();

        // Index name

        String _index = response.getIndex();

        System.out.println(_index);

        // Type name

        String _type = response.getType();

        System.out.println(_type);

        // Document ID (generated or not)

        String _id = response.getId();

        System.out.println(_id);

        // Version (if it's the first time you index this document, you will get: 1)

        long _version = response.getVersion();

        System.out.println(_version);

        // isCreated() is true if the document is a new one, false if it has been updated

        boolean created = response.isCreated();

        System.out.println(created);

    }

     

    @Test

    //更新操作，为既有的index添加一列

    public void testUpdateIndex() throws IOException, InterruptedException, ExecutionException{

        UpdateRequest updateRequest = new UpdateRequest("bank2", "account2", "AU_04YOqvDFJcKoDMaDT")

        .doc(XContentFactory.jsonBuilder()

            .startObject()

                .field("gender", "male")

            .endObject());

        client.update(updateRequest).get();

        /*IndexRequest indexRequest = new IndexRequest("bank2", "account2", "2")

            .source(XContentFactory.jsonBuilder()

            .startObject()

                .field("name", "Joe Smith")

                //.field("gender", "male")

            .endObject());

        UpdateRequest updateRequest = new UpdateRequest("bank2", "account2", "2")

            .doc(XContentFactory.jsonBuilder()

            .startObject()

                .field("gender", "female")

            .endObject())

        .upsert(indexRequest);             

        client.update(updateRequest).get();*/

    }

     

    @Test

    //查询

    public void testSearchIndex(){

        SearchResponse response = client.prepareSearch("bankfortest")

                .setTypes("account")

                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)

                .setQuery(QueryBuilders.termQuery("lastname", "fox")) // Query

                .setFrom(0).setSize(10).setExplain(true).execute().actionGet();

        SearchHits hits = response.getHits();

        System.out.println(hits.getTotalHits());

        for (int i = 0; i < hits.getHits().length; i++) {

            System.out.println(hits.getHits()[i].getSourceAsString());

        }

    }

 

    @Test

    //删除一条记录

    public void testDeleteIndex(){

        DeleteResponse response = client.prepareDelete("bank1", "account1", "AU_02Z00vDFJcKoDMaDS")

                .execute()

                .actionGet();

    }

}