ElasticSerach初探第一篇 - 认识ES+环境搭建+简单MySQL数据同步+SpringBoot整合ES

一、认识ElasticSearch

是一个基于Lucene的开源搜索引擎，通过简单的RESTful API来隐藏Lucene的复杂性。全文搜索，分析系统，分布式数据库；elasticsearch的革命性就在于将这些单一的有用的技术整合成一个一体化的”实时的应用”。

 

二、ElasticSearch环境搭建

1.安装ElasticSearch

从官网下载，https://www.elastic.co/downloads/elasticsearch，选择自己想要的版本下载到本地，然后解压；

 

2.安装Kibana

从官网下载，https://www.elastic.co/downloads/kibana，这里选择和ElasticSearch一样的版本，然后解压。

Kibana是一个为 ElasticSearch 提供数据分析的 Web 接口。可使用它对日志进行高效的搜索、可视化、分析等各种操作。

 

3.安装X-pack

是ElasticSearch的一个扩展包，包括安全、警告、监视、图形、报告等功能集成在一个软件包，方便我们操作ElasticSearch。

到ElasticSearch目录，执行./bin/elasticsearch-plugin install x-pack；到Kibana目录，执行./bin/kibana-plugin install x-pack。

 

4.启动ElasticSearch和Kibana+访问

启动ElasticSearch：ES目录执行./bin/elasticsearch；

启动Kibana：Kibana目录执行./bin/kibana

 

访问Kibana：http://localhost:5601/

 

 

特别注意：登录的默认用户名和密码分别是：elastic和changeme 。

 

三、MySQL数据同步到ElasticSearch

自己在网上找了很多资料，最后选择采用logstash-input-jdbc的方式进行MySQL数据到ES的同步，已经做了一个初步的尝试，感觉还不错。现在仅仅尝试了一次同步一个表的数据，批量数据同步还需要研究。下面是我们进行数据同步的一些步骤：

 

步骤一：安装Ruby

因为logstash-input-jdbc是logstash的一个插件，使用ruby开发。自己本地是在MacOS上，安装方式：ruby -e "$(curl --insecure -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"；

 

步骤二：安装logstash-input-jdbc

首先从官网下载logstash:https://www.elastic.co/downloads/logstash，解压之后进入本地的logstash目录，到bin目录下面执行命令：./logstash-plugin install logstash-input-jdbc，安装的过程需要花费一点时间，静待一会儿，耐心等待，直到出现Installation successful。

 

步骤三：使用logstash-input-jdbc

进入本地的logstash/bin目录，创建mysql目录，并添加jdbc.conf和jdbc.sql两个文件，并且加入mysql驱动。下面贴出jdbc.conf和jdbc.sql文件的内容，这里实现了同步MySQL中一个表的数据到ES。

 

jdbc.conf文件内容如下：

input {  

    stdin {

    }  

 

    jdbc {  

      # mysql 数据库链接,test为数据库名  

      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test"  

      # 登录mysql数据库用户名和密码  

      jdbc_user => "root"  

      jdbc_password => "root"  

      # 驱动  

      jdbc_driver_library => "/Users/wuhoujian/Documents/myself/learning/elasticsearch/logstash-5.5.1/bin/mysql/mysql-connector-java-5.1.38.jar"  

      # 驱动类名  

      jdbc_driver_class => "com.mysql.jdbc.Driver"  

      jdbc_paging_enabled => "true"  

      jdbc_page_size => "50000"  

      # 执行的sql 文件路径+名称  

      statement_filepath => "/Users/wuhoujian/Documents/myself/learning/elasticsearch/logstash-5.5.1/bin/mysql/jdbc.sql"  

      # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新  

      #schedule => "* * * * *"  

      # 索引类型  

      type => "pattern"  

    }  

}  

  

filter {  

    json {  

        source => "pattern" 

    }  

}  

  

output {  

    elasticsearch {  

        # ES的IP地址及端口  

        hosts => ["localhost:9200"]  

        # 索引名称  

        index => "test"  

        # 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号  

        document_id => "%{id}"  

    }  

    stdout {  

       # JSON格式输出  

        codec => json_lines  

    }  

 

}

 

jdbc.sql文件内容如下：

select * from 要同步的MySQL数据库中的表名

 

bin目录下执行命令：.／logstash -f  .／mysql／jdbc.conf

 

四、Java HTTP REST API访问ElasticSearch

Java访问ES的两种方式：第一种——通过TransportClient访问；第二种——通过Jest访问。我们采用Jest(一个封装了对ES各种操作的轻量的框架)。

 

步骤一：pom.xml中添加相关依赖

<dependency>

    <groupId>io.searchbox</groupId>

    <artifactId>jest</artifactId>

    <version>5.3.3</version>

</dependency>

 

<dependency>

<groupId>org.elasticsearch</groupId>

<artifactId>elasticsearch</artifactId>

<version>5.5.1.0</version>

</dependency>

 

<dependency>

<groupId>org.apache.httpcomponents</groupId>

<artifactId>httpcore</artifactId>

<version>4.4.7</version>

</dependency>

 

<dependency>

<groupId>org.apache.httpcomponents</groupId>

<artifactId>httpclient</artifactId>

<version>4.5.3</version>

</dependency>

 

<dependency>

<groupId>commons-logging</groupId>

<artifactId>commons-logging</artifactId>

<version>1.1.1</version>

 

</dependency>

 

步骤二：准备工具类，封装各种Jest操作ES的方法

 package com.jhzz.jizhitong.common.util;

 

import java.util.List;

 

import com.google.gson.GsonBuilder;

 

import io.searchbox.client.JestClient;

import io.searchbox.client.JestClientFactory;

import io.searchbox.client.JestResult;

import io.searchbox.client.config.HttpClientConfig;

import io.searchbox.core.Bulk;

import io.searchbox.core.BulkResult;

import io.searchbox.core.Count;

import io.searchbox.core.CountResult;

import io.searchbox.core.Delete;

import io.searchbox.core.DocumentResult;

import io.searchbox.core.Get;

import io.searchbox.core.Index;

import io.searchbox.core.Search;

import io.searchbox.core.SearchResult;

import io.searchbox.indices.CreateIndex;

import io.searchbox.indices.DeleteIndex;

import io.searchbox.indices.mapping.GetMapping;

import io.searchbox.indices.mapping.PutMapping;

 

/**

 * Jest操作ES工具类

 * 

 * @author wuhoujian

 *

 */

public class JestUtil {

/**

* 获取JestClient对象

* 

* @return

*/

public static JestClient getJestClient() {

 

JestClientFactory factory = new JestClientFactory();

factory.setHttpClientConfig(new HttpClientConfig.Builder("http://127.0.0.1:9200")

.gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'hh:mm:ss").create()).connTimeout(1500)

.readTimeout(3000).multiThreaded(true).build());

return factory.getObject();

}

 

/**

* 创建索引

* 

* @param jestClient

* @param indexName

* @return

* @throws Exception

*/

public boolean createIndex(JestClient jestClient, String indexName) throws Exception {

 

JestResult jr = jestClient.execute(new CreateIndex.Builder(indexName).build());

return jr.isSucceeded();

}

 

/**

* Put映射

* 

* @param jestClient

* @param indexName

* @param typeName

* @param source

* @return

* @throws Exception

*/

public boolean createIndexMapping(JestClient jestClient, String indexName, String typeName, String source)

throws Exception {

 

PutMapping putMapping = new PutMapping.Builder(indexName, typeName, source).build();

JestResult jr = jestClient.execute(putMapping);

return jr.isSucceeded();

}

 

/**

* Get映射

* 

* @param jestClient

* @param indexName

* @param typeName

* @return

* @throws Exception

*/

public String getIndexMapping(JestClient jestClient, String indexName, String typeName) throws Exception {

 

GetMapping getMapping = new GetMapping.Builder().addIndex(indexName).addType(typeName).build();

JestResult jr = jestClient.execute(getMapping);

return jr.getJsonString();

}

 

/**

* 索引文档

* 

* @param jestClient

* @param indexName

* @param typeName

* @param objs

* @return

* @throws Exception

*/

public boolean index(JestClient jestClient, String indexName, String typeName, List<Object> objs) throws Exception {

 

Bulk.Builder bulk = new Bulk.Builder().defaultIndex(indexName).defaultType(typeName);

for (Object obj : objs) {

Index index = new Index.Builder(obj).build();

bulk.addAction(index);

}

BulkResult br = jestClient.execute(bulk.build());

return br.isSucceeded();

}

 

/**

* 搜索文档

* 

* @param jestClient

* @param indexName

* @param typeName

* @param query

* @return

* @throws Exception

*/

public SearchResult search(JestClient jestClient, String indexName, String typeName, String query)

throws Exception {

 

Search search = new Search.Builder(query).addIndex(indexName).addType(typeName).build();

return jestClient.execute(search);

}

 

/**

* Count文档

* 

* @param jestClient

* @param indexName

* @param typeName

* @param query

* @return

* @throws Exception

*/

public Double count(JestClient jestClient, String indexName, String typeName, String query) throws Exception {

 

Count count = new Count.Builder().addIndex(indexName).addType(typeName).query(query).build();

CountResult results = jestClient.execute(count);

return results.getCount();

}

 

/**

* Get文档

* 

* @param jestClient

* @param indexName

* @param typeName

* @param id

* @return

* @throws Exception

*/

public static JestResult get(JestClient jestClient, String indexName, String typeName, String id) throws Exception {

 

Get get = new Get.Builder(indexName, id).type(typeName).build();

return jestClient.execute(get);

}

 

/**

* Delete索引

* 

* @param jestClient

* @param indexName

* @return

* @throws Exception

*/

public boolean delete(JestClient jestClient, String indexName) throws Exception {

 

JestResult jr = jestClient.execute(new DeleteIndex.Builder(indexName).build());

return jr.isSucceeded();

}

 

/**

* Delete文档

* 

* @param jestClient

* @param indexName

* @param typeName

* @param id

* @return

* @throws Exception

*/

public boolean delete(JestClient jestClient, String indexName, String typeName, String id) throws Exception {

 

DocumentResult dr = jestClient.execute(new Delete.Builder(id).index(indexName).type(typeName).build());

return dr.isSucceeded();

}

}

 

 

步骤三：编码访问ES

JestResult jestResult = JestUtil.get(JestUtil.getJestClient(), "jizhitong", "pattern", "5");

 

System.out.println(jestResult.getSourceAsString());