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());
相关推荐
另外一部分，则需要先做聚类、分类处理，将聚合出的分类结果存入ES集群的聚类索引中。数据处理层的聚合结果存入ES中的指定索引，同时将每个聚合主题相关的数据存入每个document下面的某个field下。