一、引言

Elasticsearch 是一个开源的分布式搜索和分析引擎，基于 Apache Lucene 构建。它提供了一个分布式、多租户的全文搜索引擎，具有高可用性、可扩展性和高性能等特点。Elasticsearch 广泛应用于日志分析、全文搜索、商业智能等领域。

二、基本概念

1.索引（Index）

定义：索引是 Elasticsearch 中存储数据的逻辑容器，类似于关系型数据库中的数据库。一个索引可以包含多个文档，并且可以在多个节点上进行分片存储。
示例：假设我们要构建一个电商平台的搜索系统，可以创建一个名为 products 的索引来存储所有商品信息。

2.类型（Type）

定义：在 Elasticsearch 7.x 及以前的版本中，类型是索引内部的逻辑分区，类似于关系型数据库中的表。但从 Elasticsearch 8.x 开始，类型已被弃用，一个索引中只能有一个 “_doc” 类型。
历史示例（7.x 及以前）：在 products 索引中，可以定义 clothes、electronics 等不同类型来区分不同类别的商品。

3.文档（Document）

定义：文档是 Elasticsearch 中存储的基本数据单元，类似于关系型数据库中的一行记录。文档以 JSON 格式表示，包含一个或多个字段。
示例：一个商品文档可能如下所示：

{"id": 1,"name": "iPhone 15","price": 999,"category": "electronics"
}

4.分片（Shard）

定义：为了实现分布式存储和处理，索引会被划分为多个分片。每个分片是一个独立的 Lucene 索引，可以存储部分文档数据。分片可以分布在不同的节点上，提高系统的可扩展性和容错性。
示例：如果我们将 products 索引划分为 5 个分片，那么每个分片将存储一部分商品文档。

5.副本（Replica）

定义：副本是分片的复制，用于提高数据的可用性和容错性。每个分片可以有零个或多个副本，副本会分布在不同的节点上。当某个节点出现故障时，副本可以接替该节点上的分片继续提供服务。
示例：如果我们为 products 索引的每个分片设置 1 个副本，那么总共会有 10 个分片（5 个主分片和 5 个副本分片）。

二、ELK搭建

ELK分别是Elasticsearch、Logstash、Kibana

• Elasticsearch:存储大量日志，并为其创建索引，提供查询。

• Logstash：负责从应用，也就是我们的程序的日志采集、处理加工数据给Elasticsearch

• Kbana：从Elasticsearch搜索日志，并展示到页面，供用户使用。

1.创建挂载的文件

es挂载：

mkdir -p /usr/docker/elk/elasticsearch/{config,plugins,data,logs}

kibana挂载：

mkdir -p /usr/docker/elk/kibana/config

logstash挂载：

mkdir -p /usr/docker/elk/logstash/config

2.创建配置文件

ES挂载具体配置

vim /usr/docker/elk/elasticsearch/config/elasticsearch.yml
#输入下面命令：
http.host: 0.0.0.0
xpack.security.enabled: false

http.host：任何地址都可以访问。
xpack.security.enabled：关闭密码认证

Kibana挂载具体配置

vim /usr/docker/elk/kibana/config/kibana.yml
# 内容
server.host: "0.0.0.0" 
elasticsearch.url: "http://192.168.1.37:9200"  # 注意7.X以上版本用elasticsearch.hosts: ["http://192.168.1.37:9200"]
xpack.security.enabled: false

elasticsearch.hosts：指向es地址

Logstash挂载具体配置

vim /usr/docker/elk/logstash/config/logstash.yml
#内容：
http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.url: http://192.168.1.37:9200# 注意7.X以上版本用elasticsearch.hosts: ["http://192.168.1.37:9200"]#记录存放：
cd  /usr/docker/elk/logstash/config
touch  log
chmod 777 log

配置mysql

vim /usr/docker/elk/logstash/config/logstash.confinput {stdin {}jdbc {jdbc_connection_string => "jdbc:mysql://192.168.1.103:3306/wpx_resource?characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true&allowPublicKeyRetrieval=true"jdbc_user => "root"jdbc_password => "Abc123!_"jdbc_driver_library => "/usr/share/logstash/config/mysql-connector-java-8.0.18.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "3000"# 将 create_date 转换为时间戳statement => "SELECT resources_id, resources_name, resources_type, resources_size, resource_suffix, out_path, local_path, video_id, duration, all_parent_ids, parent_resource_id, parent_public_directory_ids, UNIX_TIMESTAMP(create_date) as create_date_timestamp, modify_date, create_by_id, create_by, modify_by_id, modify_by, is_share, is_public, tenant_id, download_num, public_resource_id, review_result, review_state, del_flag FROM resource_manage"schedule => "*/1 * * * *"use_column_value => falsetracking_column_type => "timestamp"# 使用转换后的字段名tracking_column => "create_date_timestamp"record_last_run => truejdbc_default_timezone => "Asia/Shanghai"last_run_metadata_path => "/usr/share/logstash/config/log"}
}output {elasticsearch {hosts => ["192.168.1.37:9200"]index => "resource_manage"document_id => "%{resources_id}"}stdout {codec => json_lines}
}

数据库连接配置：

• jdbc_connection_string：指定要连接的 MySQL 数据库的连接字符串，包括数据库的地址（192.168.239.131）、端口（3306）、数据库名（test）以及一些连接参数（如字符编码和时区）。
• jdbc_user 和 jdbc_password：分别是连接数据库的用户名和密码。
• jdbc_driver_library：指定 MySQL JDBC 驱动的 JAR 文件路径。
• jdbc_driver_class：指定 JDBC 驱动类的名称。

分页配置：

• jdbc_paging_enabled：启用分页查询，设置为 true 表示开启分页功能。
• jdbc_page_size：每次查询的记录数，这里设置为 300000。

查询配置：

• statement：定义要执行的 SQL 查询语句，这里是从 sys_log 表中查询指定的列。

定时任务配置：

• schedule：使用 cron 表达式指定查询的执行频率，*/1 * * * * 表示每分钟执行一次查询。

增量查询配置：

• use_column_value：是否使用列的值来跟踪数据，这里设置为 false。
• tracking_column_type：跟踪列的数据类型，这里是 timestamp 类型。
• tracking_column：指定用于跟踪数据的列名，这里是 oper_time 列。
• record_last_run：是否记录上次查询的时间，设置为 true 表示记录。
• last_run_metadata_path：指定记录上次查询时间的文件路径。

时区配置：

• jdbc_default_timezone：指定数据库的默认时区为 Asia/Shanghai。

注意需要自己下载mysql-connector-java-8.0.28.jar或者将本地的拷贝到挂载的logstash/config中

流水线指定上面的配置文件

vim /mydata/elk/logstash/config/pipelines.yml
#内容：- pipeline.id: resource_managepath.config: "/usr/share/logstash/config/logstash.conf"

最后/usr/docker/elk/logstash/config目录如下

防止保存没有修改权限，可以把上面建的文件夹和文件赋予修改权限：

chmod 777 文件名称

注意如果是6版本需要配置log4j2.properties文件

cd /usr/docker/elk/logstash/config
vim log4j2.properties
#添加内容
status = error
name = LogstashPropertiesConfig#filename = /usr/share/logstash/logsappender.rolling.type = RollingFile
appender.rolling.name = RollingFile
appender.rolling.fileName = logs/logstash.log
appender.rolling.filePattern = logs/logstash-%d{yyyy-MM-dd}-%i.log.gz
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
appender.rolling.policies.type = Policies
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.rolling.policies.time.interval = 1
appender.rolling.policies.time.modulate = true
#appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
#appender.rolling.policies.size.size=50MB
#appender.rolling.strategy.type = DefaultRolloverStrategy
#appender.rolling.strategy.max = 5appender.console.type = Console
appender.console.name = plain_console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%nappender.json_console.type = Console
appender.json_console.name = json_console
appender.json_console.layout.type = JSONLayout
appender.json_console.layout.compact = true
appender.json_console.layout.eventEol = truerootLogger.level = ${sys:ls.log.level}
rootLogger.appenderRef.rolling.ref = RollingFile
rootLogger.appenderRef.console.ref = ${sys:ls.log.format}_console

因为不同版本有不同的配置所以最好还是先不挂载进行运行。然后将里面的配置文件考出来之后修改，然后删除容器后重新挂载修改

3.运行容器

elasticsearch、kibana、logstash版本必须一致。如果使用springboot集成需要注意和spring版本统一。

1. docker compose一键搭建

在elk目录创建：

vim docker-compose.yml

内容如下：

services:elasticsearch:image: elasticsearch:6.5.4container_name: elasticsearchports:- "9200:9200"- "9300:9300"environment:- cluster.name=elasticsearch- discovery.type=single-node- "ES_JAVA_OPTS=-Xms512m -Xmx512m"volumes:- /usr/docker/elk/elasticsearch/plugins:/usr/share/elasticsearch/plugins- /usr/docker/elk/elasticsearch/data:/usr/share/elasticsearch/data- /usr/docker/elk/elasticsearch/logs:/usr/share/elasticsearch/logskibana:image: kibana:6.5.4container_name: kibanaports:- "5601:5601"depends_on:- elasticsearchenvironment:I18N_LOCALE: zh-CNvolumes:- /usr/docker/elk/kibana/config/kibana.yml:/usr/share/kibana/config/kibana.ymllogstash:image: logstash:6.5.4container_name: logstashports:- "5044:5044"volumes:- /usr/docker/elk/logstash/config:/usr/share/logstash/configdepends_on:- elasticsearch

使用docker compose 命令运行

一定要在docker-compose.yml所在目录执行命令！！

docker compose up -d

2.单独运行

运行elasticsearch

docker run --name elasticsearch -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" -e ES_JAVA_OPTS="-Xms64m -Xmx512m" -v /usr/docker/elk/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml -v /usr/docker/elk/elasticsearch/data:/usr/share/elasticsearch/data -v  /usr/docker/elk/elasticsearch/plugins:/usr/share/elasticsearch/plugins -d elasticsearch:6.5.4

运行Kibana

docker run --name kibana -v /usr/docker/elk/kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml -p 5601:5601 -d kibana:6.5.4

运行Logstash

docker run -d -p 5044:5044 -v /usr/docker/elk/logstash/config:/usr/share/logstash/config --name logstash logstash:6.5.4

三、项目集成

我这里使用的是spring-boot-starter-data-elasticsearch包集成。需要注意版本限制。

maven引入依赖

		<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId></dependency>

修改yml文件
根据版本不同有的可能不一样。需要自己查看自己版本的

  data:elasticsearch:cluster-nodes: 192.168.1.37:9200cluster-name: elasticsearch

定义映射对象

import com.baomidou.mybatisplus.extension.activerecord.Model;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;import java.math.BigDecimal;
import java.time.LocalDateTime;/*** 资源管理-es映射** @date 2025年03月13日13:15:34*/
@EqualsAndHashCode(callSuper = true)
@Data
@Document(indexName = "resource_manage", type = "doc", useServerConfiguration = true, createIndex = false)
public class EsResourceManage extends Model<EsResourceManage> {private static final long serialVersionUID = 1L;/*** id*/@Idprivate String resources_id;/*** 资源名称*/@Field(type = FieldType.Text)private String resources_name;/*** 资源类型(1:文件夹,2:视频,3:文档,4:图片,5:音频,6:模型)*/private String resources_type;/*** 资源大小*/private BigDecimal resources_size;/*** 资源后缀(doc,txt,jpg,mp4等)*/private String resource_suffix;/*** 外部存储路径*/private String out_path;}

方式一：继承 ElasticsearchRepository

ElasticsearchRepository 提供了一个高级的抽象，使得你可以在不编写任何实现代码的情况下，直接使用预定义的CRUD方法和查询方法。

/*** es映射-Repository类* @author ：lzy* @date ：2025/3/13 13:16*/
@Repository
public interface EsResourceManageRepository extends ElasticsearchRepository<EsResourceManage, String> {}

然后就可以直接使用了,属性jpa的同学可以直接上手

方式二：使用ElasticsearchRestTemplate

与 ElasticsearchRepository 相比，ElasticsearchRestTemplate 更适合用于复杂查询 。
比如 多个条件组合、范围查询、模糊查询、聚合查询等复杂场景
还支持分页、排序、过滤等高级功能

@RestController
@RequestMapping("/employeeInfo")
public class EmployeeElasticController {@Autowiredprivate ElasticsearchRestTemplate elasticsearchRestTemplate;@RequestMapping("/template/save")public String  templateSave() throws Exception  {SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");EmployeeInfo employeeInfo = new EmployeeInfo(8888L, "2001", "张八", "zhangsan", "Java", 1, 19, new BigDecimal("12500.01"), simpleDateFormat.parse("2019-09-10"), "备注");elasticsearchRestTemplate.save(employeeInfo);return "success";}

版本注意

springboot 、spring-data-elasticsearch、elasticserach的版本对应关系

对照地址