Filebeat

简介

Beats 是安装在服务器上的数据中转代理。

Beats 可以将数据直接传输到 Elasticsearch 或传输到 Logstash 。

Beats 有多种类型，可以根据实际应用需要选择合适的类型。

常用的类型有：

• Packetbeat：网络数据包分析器，提供有关您的应用程序服务器之间交换的事务的信息。
• Filebeat：从您的服务器发送日志文件。
• Metricbeat：是一个服务器监视代理程序，它定期从服务器上运行的操作系统和服务收集指标。
• Winlogbeat：提供 Windows 事件日志。

参考

更多 Beats 类型可以参考：community-beats

说明

由于本人工作中只应用了 FileBeat，所以后面内容仅介绍 FileBeat 。

FileBeat 的作用

相比 Logstash，FileBeat 更加轻量化。

在任何环境下，应用程序都有停机的可能性。 Filebeat 读取并转发日志行，如果中断，则会记住所有事件恢复联机状态时所在位置。

Filebeat 带有内部模块（auditd，Apache，Nginx，System 和 MySQL），可通过一个指定命令来简化通用日志格式的收集，解析和可视化。

FileBeat 不会让你的管道超负荷。FileBeat 如果是向 Logstash 传输数据，当 Logstash 忙于处理数据，会通知 FileBeat 放慢读取速度。一旦拥塞得到解决，FileBeat 将恢复到原来的速度并继续传播。

安装

Unix / Linux 系统建议使用下面方式安装，因为比较通用。

1
2

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.1.1-linux-x86_64.tar.gz
tar -zxf filebeat-6.1.1-linux-x86_64.tar.gz

更多内容可以参考：filebeat-installation

配置

配置文件

首先，必须要知道的是：filebeat.yml 是 filebeat 的配置文件。其路径会因为你安装方式而有所不同。

Beat 所有系列产品的配置文件都基于 YAML 格式，FileBeat 当然也不例外。

更多 filebeat 配置内容可以参考：配置 filebeat

更多 filebeat.yml 文件格式内容可以参考：filebeat.yml 文件格式

filebeat.yml 部分配置示例：

1
2
3
4
5
6
7
8

filebeat:
  prospectors:
    - type: log
      paths:
        - /var/log/*.log
      multiline:
        pattern: '^['
        match: after

参考

更多 filebeat 配置内容可以参考：配置 filebeat

更多 filebeat.yml 文件格式内容可以参考：filebeat.yml 文件格式

重要配置项

下面我将列举 Filebeat 的较为重要的配置项。

如果想了解更多配置信息，可以参考：

更多 filebeat 配置内容可以参考：配置 filebeat

更多 filebeat.yml 文件格式内容可以参考：filebeat.yml 文件格式

filebeat.prospectors

（文件监视器）用于指定需要关注的文件。

示例

1
2
3
4
5

filebeat.prospectors:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log

output.elasticsearch

如果你希望使用 filebeat 直接向 elasticsearch 输出数据，需要配置 output.elasticsearch 。

示例

1
2

output.elasticsearch:
  hosts: ['192.168.1.42:9200']

output.logstash

如果你希望使用 filebeat 向 logstash 输出数据，然后由 logstash 再向 elasticsearch 输出数据，需要配置 output.logstash。

注意

相比于向 elasticsearch 输出数据，个人更推荐向 logstash 输出数据。

因为 logstash 和 filebeat 一起工作时，如果 logstash 忙于处理数据，会通知 FileBeat 放慢读取速度。一旦拥塞得到解决，FileBeat 将恢复到原来的速度并继续传播。这样，可以减少管道超负荷的情况。

示例

1
2

output.logstash:
  hosts: ['127.0.0.1:5044']

此外，还需要在 logstash 的配置文件（如 logstash.conf）中指定 beats input 插件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

input {
  beats {
    port => 5044 # 此端口需要与 filebeat.yml 中的端口相同
  }
}

# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }

output {
  elasticsearch {
    hosts => "localhost:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
}

setup.kibana

如果打算使用 Filebeat 提供的 Kibana 仪表板，需要配置 setup.kibana 。

示例

1
2

setup.kibana:
  host: 'localhost:5601'

setup.template.settings

在 Elasticsearch 中，索引模板用于定义设置和映射，以确定如何分析字段。

在 Filebeat 中，setup.template.settings 用于配置索引模板。

Filebeat 推荐的索引模板文件由 Filebeat 软件包安装。如果您接受 filebeat.yml 配置文件中的默认配置，Filebeat 在成功连接到 Elasticsearch 后自动加载模板。

您可以通过在 Filebeat 配置文件中配置模板加载选项来禁用自动模板加载，或加载自己的模板。您还可以设置选项来更改索引和索引模板的名称。

参考

更多内容可以参考：filebeat-template

说明

如无必要，使用 Filebeat 配置文件中的默认索引模板即可。

setup.dashboards

Filebeat 附带了示例 Kibana 仪表板。在使用仪表板之前，您需要创建索引模式 filebeat- *，并将仪表板加载到 Kibana 中。为此，您可以运行 setup 命令或在 filebeat.yml 配置文件中配置仪表板加载。

为了在 Kibana 中加载 Filebeat 的仪表盘，需要在 filebeat.yml 配置中启动开关：

1

setup.dashboards.enabled: true

参考

更多内容可以参考：configuration-dashboards

命令

filebeat 提供了一系列命令来完成各种功能。

执行命令方式：

1

./filebeat COMMAND

参考

更多内容可以参考：command-line-options

说明

个人认为命令行没有必要一一掌握，因为绝大部分功能都可以通过配置来完成。且通过命令行指定功能这种方式要求每次输入同样参数，不利于固化启动方式。

最重要的当然是启动命令 run 了。

示例 指定配置文件启动

1 2	./filebeat run -e -c filebeat.yml -d "publish" ./filebeat -e -c filebeat.yml -d "publish" # run 可以省略

模块

Filebeat 和 Metricbeat 内部集成了一系列模块，用以简化常见日志格式（例如 NGINX、Apache 或诸如 Redis 或 Docker 等系统指标）的收集、解析和可视化过程。

• 配置 elasticsearch 和 kibana

1
2
3
4
5
6
7
8

output.elasticsearch:
  hosts: ["myEShost:9200"]
  username: "elastic"
  password: "elastic"
setup.kibana:
  host: "mykibanahost:5601"
  username: "elastic"
  password: "elastic

username 和 password 是可选的，如果不需要认证则不填。

• 初始化环境

执行下面命令，filebeat 会加载推荐索引模板。

1

./filebeat setup -e

• 指定模块

执行下面命令，指定希望加载的模块。

1

./filebeat -e --modules system,nginx,mysql

更多内容可以参考：

• 配置 filebeat 模块
• filebeat 支持模块

原理

Filebeat 有两个主要组件：

harvester：负责读取一个文件的内容。它会逐行读取文件内容，并将内容发送到输出目的地。

prospector：负责管理 harvester 并找到所有需要读取的文件源。比如类型是日志，prospector 就会遍历制定路径下的所有匹配要求的文件。

1
2
3
4
5

filebeat.prospectors:
  - type: log
    paths:
      - /var/log/*.log
      - /var/path2/*.log

Filebeat 保持每个文件的状态，并经常刷新注册表文件中的磁盘状态。状态用于记住 harvester 正在读取的最后偏移量，并确保发送所有日志行。

Filebeat 将每个事件的传递状态存储在注册表文件中。所以它能保证事件至少传递一次到配置的输出，没有数据丢失。

参考资料

• Beats 官网
• Beats Github
• Beats 官方文档

Kibana

通过 Kibana，您可以对自己的 Elasticsearch 进行可视化，还可以在 Elastic Stack 中进行导航，这样您便可以进行各种操作了，从跟踪查询负载，到理解请求如何流经您的整个应用，都能轻松完成。

安装

环境要求

版本：Elastic Stack 7.4

安装步骤

安装步骤如下：

1. 在 kibana 官方下载地址 下载所需版本包并解压到本地。
2. 修改 config/kibana.yml 配置文件，设置 elasticsearch.url 指向 Elasticsearch 实例。
3. 运行 bin/kibana （Windows 上运行 bin\kibana.bat）
4. 在浏览器上访问 http://localhost:5601

使用

检索

单击侧面导航栏中的 检索（Discover） ，可以显示 Kibana 的数据查询功能功能。

在搜索栏中，您可以输入 Elasticsearch 查询条件来搜索您的数据。您可以在 Discover 页面中浏览结果并在 Visualize 页面中创建已保存搜索条件的可视化。

当前索引模式显示在查询栏下方。索引模式确定提交查询时搜索哪些索引。要搜索一组不同的索引，请从下拉菜单中选择不同的模式。要添加索引模式（index pattern），请转至 Management/Kibana/Index Patterns 并单击 Add New。

您可以使用字段名称和您感兴趣的值构建搜索。对于数字字段，可以使用比较运算符，如大于（>），小于（<）或等于（=）。您可以将元素与逻辑运算符 AND，OR 和 NOT 链接，全部使用大写。

默认情况下，每个匹配文档都显示所有字段。要选择要显示的文档字段，请将鼠标悬停在“可用字段”列表上，然后单击要包含的每个字段旁边的添加按钮。例如，如果只添加 account_number，则显示将更改为包含五个帐号的简单列表：

kibana 的搜索栏遵循 query-string-syntax 文档中所说明的查询语义。

这里说明一些最基本的查询语义。

查询字符串会被解析为一系列的术语和运算符。一个术语可以是一个单词（如：quick、brown）或用双引号包围的短语（如”quick brown”）。

查询操作允许您自定义搜索 - 下面介绍了可用的选项。

2.1.1. 字段名称

正如查询字符串查询中所述，将在搜索条件中搜索 default_field，但可以在查询语法中指定其他字段：

例如：

• 查询 status 字段中包含 active 关键字

1

status:active

• title 字段包含 quick 或 brown 关键字。如果您省略 OR 运算符，则将使用默认运算符

1
2

title:(quick OR brown)
title:(quick brown)

• author 字段查找精确的短语 “john smith”，即精确查找。

1

author:"John Smith"

• 任意字段 book.title，book.content 或 book.date 都包含 quick 或 brown（注意我们需要如何使用 \* 表示通配符）

1

book.\*:(quick brown)

• title 字段包含任意非 null 值

1

_exists_:title

2.1.2. 通配符

ELK 提供了 ? 和 * 两个通配符。

• ? 表示任意单个字符；
• * 表示任意零个或多个字符。

1

qu?ck bro*

注意：通配符查询会使用大量的内存并且执行性能较为糟糕，所以请慎用。 > 提示：纯通配符 * 被写入 exsits 查询，从而提高了查询效率。因此，通配符 field：* 将匹配包含空值的文档，如：{“field”：“”}，但是如果字段丢失或显示将值置为 null 则不匹配，如：“field”：null} > 提示：在一个单词的开头（例如：*ing）使用通配符这种方式的查询量特别大，因为索引中的所有术语都需要检查，以防万一匹配。通过将 allow_leading_wildcard 设置为 false，可以禁用。

2.1.3. 正则表达式

可以通过 / 将正则表达式包裹在查询字符串中进行查询

例：

1

name:/joh?n(ath[oa]n)/

支持的正则表达式语义可以参考：Regular expression syntax

2.1.4. 模糊查询

我们可以使用 ~ 运算符来进行模糊查询。

例：

假设我们实际想查询

1

quick brown forks

但是，由于拼写错误，我们的查询关键字变成如下情况，依然可以查到想要的结果。

1

quikc\~ brwn\~ foks\~

这种模糊查询使用 Damerau-Levenshtein 距离来查找所有匹配最多两个更改的项。所谓的更改是指单个字符的插入，删除或替换，或者两个相邻字符的换位。

默认编辑距离为 2，但编辑距离为 1 应足以捕捉所有人类拼写错误的 80％。它可以被指定为：

1

quikc\~1

2.1.5. 近似检索

尽管短语查询（例如，john smith）期望所有的词条都是完全相同的顺序，但是近似查询允许指定的单词进一步分开或以不同的顺序排列。与模糊查询可以为单词中的字符指定最大编辑距离一样，近似搜索也允许我们指定短语中单词的最大编辑距离：

例

1

"fox quick"\~5

字段中的文本越接近查询字符串中指定的原始顺序，该文档就越被认为是相关的。当与上面的示例查询相比时，短语 "quick fox" 将被认为比 "quick brown fox" 更近似查询条件。

2.1.6. 范围

可以为日期，数字或字符串字段指定范围。闭区间范围用方括号 [min TO max] 和开区间范围用花括号 {min TO max} 来指定。

我们不妨来看一些示例。

• 2012 年的所有日子

1

date:[2012-01-01 TO 2012-12-31]

• 数字 1 到 5

1

count:[1 TO 5]

• 在 alpha 和 omega 之间的标签，不包括 alpha 和 omega

1

tags:{alpha TO omega}

• 10 以上的数字

1

count:[10 TO *]

• 2012 年以前的所有日期

1

date:{* TO 2012-01-01}

此外，开区间和闭区间也可以组合使用

• 数组 1 到 5，但不包括 5

1

count:[1 TO 5}

一边无界的范围也可以使用以下语法：

1
2
3
4

age:>10
age:>=10
age:<10
age:<=10

当然，你也可以使用 AND 运算符来得到连个查询结果的交集

1
2

age:(>=10 AND <20)
age:(+>=10 +<20)

2.1.7. Boosting

使用操作符 ^ 使一个术语比另一个术语更相关。例如，如果我们想查找所有有关狐狸的文档，但我们对狐狸特别感兴趣：

1

quick^2 fox

默认提升值是 1，但可以是任何正浮点数。 0 到 1 之间的提升减少了相关性。

增强也可以应用于短语或组：

1

"john smith"^2   (foo bar)^4

2.1.8. 布尔操作

默认情况下，只要一个词匹配，所有词都是可选的。搜索 foo bar baz 将查找包含 foo 或 bar 或 baz 中的一个或多个的任何文档。我们已经讨论了上面的default_operator，它允许你强制要求所有的项，但也有布尔运算符可以在查询字符串本身中使用，以提供更多的控制。

首选的操作符是 +（此术语必须存在）和 - （此术语不得存在）。所有其他条款是可选的。例如，这个查询：

1

quick brown +fox -news

这条查询意味着：

• fox 必须存在
• news 必须不存在
• quick 和 brown 是可有可无的

熟悉的运算符 AND，OR 和 NOT（也写成 &&，|| 和 !）也被支持。然而，这些操作符有一定的优先级：NOT 优先于 AND，AND 优先于 OR。虽然 + 和 - 仅影响运算符右侧的术语，但 AND 和 OR 会影响左侧和右侧的术语。

2.1.9. 分组

多个术语或子句可以用圆括号组合在一起，形成子查询

1

(quick OR brown) AND fox

可以使用组来定位特定的字段，或者增强子查询的结果：

1

status:(active OR pending) title:(full text search)^2

2.1.10. 保留字

如果你需要使用任何在你的查询本身中作为操作符的字符（而不是作为操作符），那么你应该用一个反斜杠来转义它们。例如，要搜索（1 + 1）= 2，您需要将查询写为 \(1\+1\)\=2

保留字符是：+ - = && || > < ! ( ) { } [ ] ^ " ~ * ? : \ /

无法正确地转义这些特殊字符可能会导致语法错误，从而阻止您的查询运行。

2.1.11. 空查询

如果查询字符串为空或仅包含空格，则查询将生成一个空的结果集。

可视化

要想使用可视化的方式展示您的数据，请单击侧面导航栏中的 可视化（Visualize）。

Visualize 工具使您能够以多种方式（如饼图、柱状图、曲线图、分布图等）查看数据。要开始使用，请点击蓝色的 Create a visualization 或 + 按钮。

有许多可视化类型可供选择。

下面，我们来看创建几个图标示例：

2.2.1. Pie

您可以从保存的搜索中构建可视化文件，也可以输入新的搜索条件。要输入新的搜索条件，首先需要选择一个索引模式来指定要搜索的索引。

默认搜索匹配所有文档。最初，一个“切片”包含整个饼图：

要指定在图表中展示哪些数据，请使用 Elasticsearch 存储桶聚合。分组汇总只是将与您的搜索条件相匹配的文档分类到不同的分类中，也称为分组。

为每个范围定义一个存储桶：

1. 单击 Split Slices。
2. 在 Aggregation 列表中选择 Terms。_注意：这里的 Terms 是 Elk 采集数据时定义好的字段或标签_。
3. 在 Field 列表中选择 level.keyword。
4. 点击 按钮来更新图表。

完成后，如果想要保存这个图表，可以点击页面最上方一栏中的 Save 按钮。

2.2.2. Vertical Bar

我们在展示一下如何创建柱状图。

1. 点击蓝色的 Create a visualization 或 + 按钮。选择 Vertical Bar
2. 选择索引模式。由于您尚未定义任何 bucket ，因此您会看到一个大栏，显示与默认通配符查询匹配的文档总数。
3. 指定 Y 轴所代表的字段
4. 指定 X 轴所代表的字段
5. 点击 按钮来更新图表。

完成后，如果想要保存这个图表，可以点击页面最上方一栏中的 Save 按钮。

报表

报表（Dashboard） 可以整合和共享 Visualize 集合。

1. 点击侧面导航栏中的 Dashboard。
2. 点击添加显示保存的可视化列表。
3. 点击之前保存的 Visualize，然后点击列表底部的小向上箭头关闭可视化列表。
4. 将鼠标悬停在可视化对象上会显示允许您编辑，移动，删除和调整可视化对象大小的容器控件。

FAQ

Kibana No Default Index Pattern Warning

问题：安装 ELK 后，访问 kibana 页面时，提示以下错误信息：

1
2
3

Warning No default index pattern. You must select or create one to continue.
...
Unable to fetch mapping. Do you have indices matching the pattern?

这就说明 logstash 没有把日志写入到 elasticsearch。

解决方法：

检查 logstash 与 elasticsearch 之间的通讯是否有问题，一般问题就出在这。

参考资料

• Kibana 官网
• Kibana Github
• Kibana 官方文档

Logstash

简介

Logstash 是开源的服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的“存储库”中。

功能

Logstash 是 Elasticsearch 的最佳数据管道。

Logstash 是插件式管理模式，在输入、过滤、输出以及编码过程中都可以使用插件进行定制。Logstash 社区有超过 200 种可用插件。

工作原理

Logstash 有两个必要元素：input 和 output ，一个可选元素：filter。

这三个元素，分别代表 Logstash 事件处理的三个阶段：输入 > 过滤器 > 输出。

• input 负责从数据源采集数据。
• filter 将数据修改为你指定的格式或内容。
• output 将数据传输到目的地。

在实际应用场景中，通常输入、输出、过滤器不止一个。Logstash 的这三个元素都使用插件式管理方式，用户可以根据应用需要，灵活的选用各阶段需要的插件，并组合使用。

后面将对插件展开讲解，暂且不表。

安装

安装步骤

安装步骤如下：

（1）在 logstash 官方下载地址 下载所需版本包并解压到本地。

（2）添加一个 logstash.conf 文件，指定要使用的插件以及每个插件的设置。举个简单的例子：

1
2
3
4
5

input { stdin { } }
output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

（3）运行 bin/logstash -f logstash.conf （Windows 上运行bin/logstash.bat -f logstash.conf）

配置

设置文件

• **logstash.yml**：logstash 的默认启动配置文件
• **jvm.options**：logstash 的 JVM 配置文件。
• startup.options (Linux)：包含系统安装脚本在 /usr/share/logstash/bin 中使用的选项为您的系统构建适当的启动脚本。安装 Logstash 软件包时，系统安装脚本将在安装过程结束时执行，并使用 startup.options 中指定的设置来设置用户，组，服务名称和服务描述等选项。

logstash.yml 设置项

节选部分设置项，更多项请参考：https://www.elastic.co/guide/en/logstash/current/logstash-settings-file.html

启动

命令行

通过命令行启动 logstash 的方式如下：

1

bin/logstash [options]

其中 options 是您可以指定用于控制 Logstash 执行的命令行标志。

在命令行上设置的任何标志都会覆盖 Logstash 设置文件（logstash.yml）中的相应设置，但设置文件本身不会更改。

注

虽然可以通过指定命令行参数的方式，来控制 logstash 的运行方式，但显然这么做很麻烦。

建议通过指定配置文件的方式，来控制 logstash 运行，启动命令如下：

1	bin/logstash -f logstash.conf

若想了解更多的命令行参数细节，请参考：https://www.elastic.co/guide/en/logstash/current/running-logstash-command-line.html

配置文件

上节，我们了解到，logstash 可以执行 bin/logstash -f logstash.conf ，按照配置文件中的参数去覆盖默认设置文件（logstash.yml）中的设置。

这节，我们就来学习一下这个配置文件如何配置参数。

配置文件结构

在工作原理一节中，我们已经知道了 Logstash 主要有三个工作阶段 input 、filter、output。而 logstash 配置文件文件结构也与之相对应：

1
2
3
4
5

input {}

filter {}

output {}

每个部分都包含一个或多个插件的配置选项。如果指定了多个过滤器，则会按照它们在配置文件中的显示顺序应用它们。

插件配置

插件的配置由插件名称和插件的一个设置块组成。

下面的例子中配置了两个输入文件配置：

1
2
3
4
5
6
7
8
9
10
11

input {
  file {
    path => "/var/log/messages"
    type => "syslog"
  }

  file {
    path => "/var/log/apache/access.log"
    type => "apache"
  }
}

您可以配置的设置因插件类型而异。你可以参考： Input Plugins, Output Plugins, Filter Plugins, 和 Codec Plugins 。

值类型

一个插件可以要求设置的值是一个特定的类型，比如布尔值，列表或哈希值。以下值类型受支持。

• Array

1

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

• Lists

1
2

;(path) => ['/var/log/messages', '/var/log/*.log']
;(uris) => ['http://elastic.co', 'http://example.net']

• Boolean

1

;(ssl_enable) => true

• Bytes

1
2
3
4

my_bytes => "1113"   # 1113 bytes
my_bytes => "10MiB"  # 10485760 bytes
my_bytes => "100kib" # 102400 bytes
my_bytes => "180 mb" # 180000000 bytes

• Codec

1

;(codec) => 'json'

• Hash

1
2
3
4
5

match => {
  "field1" => "value1"
  "field2" => "value2"
  ...
}

• Number

1

;(port) => 33

• Password

1

;(my_password) => 'password'

• URI

1

;(my_uri) => 'http://foo:bar@example.net'

• Path

1

;(my_path) => '/tmp/logstash'

• String

• 转义字符

插件

input

Logstash 支持各种输入选择 ，可以在同一时间从众多常用来源捕捉事件。能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。

常用 input 插件

• file：从文件系统上的文件读取，就像 UNIX 命令 tail -0F 一样
• syslog：在众所周知的端口 514 上侦听系统日志消息，并根据 RFC3164 格式进行解析
• redis：从 redis 服务器读取，使用 redis 通道和 redis 列表。 Redis 经常用作集中式 Logstash 安装中的“代理”，它将来自远程 Logstash“托运人”的 Logstash 事件排队。
• beats：处理由 Filebeat 发送的事件。

更多详情请见：Input Plugins

filter

过滤器是 Logstash 管道中的中间处理设备。如果符合特定条件，您可以将条件过滤器组合在一起，对事件执行操作。

常用 filter 插件

• grok：解析和结构任意文本。 Grok 目前是 Logstash 中将非结构化日志数据解析为结构化和可查询的最佳方法。
• mutate：对事件字段执行一般转换。您可以重命名，删除，替换和修改事件中的字段。
• drop：完全放弃一个事件，例如调试事件。
• clone：制作一个事件的副本，可能会添加或删除字段。
• geoip：添加有关 IP 地址的地理位置的信息（也可以在 Kibana 中显示惊人的图表！）

更多详情请见：Filter Plugins

output

输出是 Logstash 管道的最后阶段。一个事件可以通过多个输出，但是一旦所有输出处理完成，事件就完成了执行。

常用 output 插件

• elasticsearch：将事件数据发送给 Elasticsearch（推荐模式）。
• file：将事件数据写入文件或磁盘。
• graphite：将事件数据发送给 graphite（一个流行的开源工具，存储和绘制指标。 http://graphite.readthedocs.io/en/latest/）。
• statsd：将事件数据发送到 statsd （这是一种侦听统计数据的服务，如计数器和定时器，通过 UDP 发送并将聚合发送到一个或多个可插入的后端服务）。

更多详情请见：Output Plugins

codec

用于格式化对应的内容。

常用 codec 插件

• json：以 JSON 格式对数据进行编码或解码。
• multiline：将多行文本事件（如 java 异常和堆栈跟踪消息）合并为单个事件。

更多插件请见：Codec Plugins

实战

前面的内容都是对 Logstash 的介绍和原理说明。接下来，我们来实战一些常见的应用场景。

传输控制台数据

stdin input 插件从标准输入读取事件。这是最简单的 input 插件，一般用于测试场景。

应用

（1）创建 logstash-input-stdin.conf ：

1
2
3
4
5

input { stdin { } }
output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

更多配置项可以参考：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-stdin.html

（2）执行 logstash，使用 -f 来指定你的配置文件：

1

bin/logstash -f logstash-input-stdin.conf

传输 logback 日志

elk 默认使用的 Java 日志工具是 log4j2 ，并不支持 logback 和 log4j。

想使用 logback + logstash ，可以使用 logstash-logback-encoder 。logstash-logback-encoder 提供了 UDP / TCP / 异步方式来传输日志数据到 logstash。

如果你使用的是 log4j ，也不是不可以用这种方式，只要引入桥接 jar 包即可。如果你对 log4j 、logback ，或是桥接 jar 包不太了解，可以参考我的这篇博文：细说 Java 主流日志工具库 。

TCP 应用

logstash 配置

（1）创建 logstash-input-tcp.conf ：

1
2
3
4
5
6
7
8
9
10

input {
tcp {
  port => 9251
  codec => json_lines
  mode => server
}
output {
 elasticsearch { hosts => ["localhost:9200"] }
 stdout { codec => rubydebug }
}

更多配置项可以参考：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-tcp.html

（2）执行 logstash，使用 -f 来指定你的配置文件：bin/logstash -f logstash-input-udp.conf

java 应用配置

（1）在 Java 应用的 pom.xml 中引入 jar 包：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

<dependency>
 <groupId>net.logstash.logback</groupId>
 <artifactId>logstash-logback-encoder</artifactId>
 <version>4.11</version>
</dependency>

<!-- logback 依赖包 -->
<dependency>
 <groupId>ch.qos.logback</groupId>
 <artifactId>logback-core</artifactId>
 <version>1.2.3</version>
</dependency>
<dependency>
 <groupId>ch.qos.logback</groupId>
 <artifactId>logback-classic</artifactId>
 <version>1.2.3</version>
</dependency>
<dependency>
 <groupId>ch.qos.logback</groupId>
 <artifactId>logback-access</artifactId>
 <version>1.2.3</version>
</dependency>

（2）接着，在 logback.xml 中添加 appender

1
2
3
4
5
6
7
8
9
10
11

<appender name="ELK-TCP" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
 <!--
 destination 是 logstash 服务的 host:port，
 相当于和 logstash 建立了管道，将日志数据定向传输到 logstash
 -->
 <destination>192.168.28.32:9251</destination>
 <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"/>
</appender>
<logger name="io.github.dunwu.spring" level="TRACE" additivity="false">
 <appender-ref ref="ELK-TCP" />
</logger>

（3）接下来，就是 logback 的具体使用 ，如果对此不了解，不妨参考一下我的这篇博文：细说 Java 主流日志工具库 。

实例：我的 logback.xml

UDP 应用

UDP 和 TCP 的使用方式大同小异。

logstash 配置

（1）创建 logstash-input-udp.conf ：

1
2
3
4
5
6
7
8
9

input {
udp {
  port => 9250
  codec => json
}
output {
 elasticsearch { hosts => ["localhost:9200"] }
 stdout { codec => rubydebug }
}

更多配置项可以参考：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-udp.html

（2）执行 logstash，使用 -f 来指定你的配置文件：bin/logstash -f logstash-input-udp.conf

java 应用配置

（1）在 Java 应用的 pom.xml 中引入 jar 包：

与 TCP 应用 一节中的引入依赖包完全相同。

（2）接着，在 logback.xml 中添加 appender

1
2
3
4
5
6
7

<appender name="ELK-UDP" class="net.logstash.logback.appender.LogstashSocketAppender">
  <host>192.168.28.32</host>
  <port>9250</port>
</appender>
<logger name="io.github.dunwu.spring" level="TRACE" additivity="false">
  <appender-ref ref="ELK-UDP" />
</logger>

（3）接下来，就是 logback 的具体使用 ，如果对此不了解，不妨参考一下我的这篇博文：细说 Java 主流日志工具库 。

实例：我的 logback.xml

传输文件

在 Java Web 领域，需要用到一些重要的工具，例如 Tomcat 、Nginx 、Mysql 等。这些不属于业务应用，但是它们的日志数据对于定位问题、分析统计同样很重要。这时无法使用 logback 方式将它们的日志传输到 logstash。

如何采集这些日志文件呢？别急，你可以使用 logstash 的 file input 插件。

需要注意的是，传输文件这种方式，必须在日志所在的机器上部署 logstash 。

应用

logstash 配置

（1）创建 logstash-input-file.conf ：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

input {
 file {
  path => ["/var/log/nginx/access.log"]
  type => "nginx-access-log"
  start_position => "beginning"
 }
}

output {
 if [type] == "nginx-access-log" {
  elasticsearch {
   hosts => ["localhost:9200"]
   index => "nginx-access-log"
  }
 }
}

（2）执行 logstash，使用 -f 来指定你的配置文件：bin/logstash -f logstash-input-file.conf

更多配置项可以参考：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html

小技巧

启动、终止应用

如果你的 logstash 每次都是通过指定配置文件方式启动。不妨建立一个启动脚本。

1
2

# cd xxx 进入 logstash 安装目录下的 bin 目录
logstash -f logstash.conf

如果你的 logstash 运行在 linux 系统下，不妨使用 nohup 来启动一个守护进程。这样做的好处在于，即使关闭终端，应用仍会运行。

创建 startup.sh

1

nohup ./logstash -f logstash.conf >> nohup.out 2>&1 &

终止应用没有什么好方法，你只能使用 ps -ef | grep logstash ，查出进程，将其 kill 。不过，我们可以写一个脚本来干这件事：

创建 shutdown.sh

脚本不多解释，请自行领会作用。

1
2

PID=`ps -ef | grep logstash | awk '{ print $2}' | head -n 1`
kill -9 ${PID}

参考资料

• Logstash 官网
• Logstash Github
• Logstash 官方文档
• logstash-logback-encoder
• ELK Stack 权威指南
• ELK（Elasticsearch、Logstash、Kibana）安装和配置

ElasticSearch API

Elasticsearch 是一个分布式、RESTful 风格的搜索和数据分析引擎，能够解决不断涌现出的各种用例。 作为 Elastic Stack 的核心，它集中存储您的数据，帮助您发现意料之中以及意料之外的情况。

Elasticsearch 基于搜索库 Lucene 开发。ElasticSearch 隐藏了 Lucene 的复杂性，提供了简单易用的 REST API / Java API 接口（另外还有其他语言的 API 接口）。

_以下简称 ES_。

REST API 最详尽的文档应该参考：ES 官方 REST API

ElasticSearch API 简介

Elasticsearch 官方提供了很多版本的 Java 客户端，包含但不限于：

• Transport Client - 7.0 废弃，8.0 移除。
• Java REST 客户端
• Elasticsearch Java API Client -

如果当前是：8.X 版本，推荐 Elasticsearch Java API客户端。

如果当前是：7.X 版本且不考虑升级，推荐 High Level REST客户端。

如果当前是：5.X、6.X 版本，推荐尽早升级集群版本。

Elasticsearch Java API Client 快速入门

:::detail 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14

//创建一个低级的客户端
final RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
//创建 JSON 对象映射器
final RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
//创建 API 客户端
final ElasticsearchClient client = new ElasticsearchClient(transport);
//查询所有索引-------------------------------------------------------------------------------------
final GetIndexResponse response = client.indices().get(query -> query.index("_all"));
final IndexState products = response.result().get("products");
System.out.println(products.toString());
//关闭
client.shutdown();
transport.close();
restClient.close();

:::

Transport Client 快速入门

TransportClient 使用 transport 模块远程连接到 Elasticsearch 集群。它不会加入集群，而只是获取一个或多个初始传输地址，并以轮询方式与它们通信。

扩展：https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/transport-client.html

:::detail 示例

启动客户端：

1
2
3
4
5
6
7

// 启动
TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
        .addTransportAddress(new TransportAddress(InetAddress.getByName("host1"), 9300))
        .addTransportAddress(new TransportAddress(InetAddress.getByName("host2"), 9300));

// 关闭
client.close();

配置集群名称

注意，如果使用的集群名称与 “elasticsearch” 不同，则必须设置集群名称。

1
2
3
4

Settings settings = Settings.builder()
        .put("cluster.name", "myClusterName").build();
TransportClient client = new PreBuiltTransportClient(settings);
// Add transport addresses and do something with the client...

启用 sniffing

1
2
3

Settings settings = Settings.builder()
        .put("client.transport.sniff", true).build();
TransportClient client = new PreBuiltTransportClient(settings);

:::

ElasticSearch Rest

ElasticSearch Rest API 语法格式

向 Elasticsearch 发出的请求的组成部分与其它普通的 HTTP 请求是一样的：

1

curl -X<VERB> '<PROTOCOL>://<HOST>:<PORT>/<PATH>?<QUERY_STRING>' -d '<BODY>'

• VERB：HTTP 方法，支持：GET, POST, PUT, HEAD, DELETE
• PROTOCOL：http 或者 https 协议（只有在 Elasticsearch 前面有 https 代理的时候可用）
• HOST：Elasticsearch 集群中的任何一个节点的主机名，如果是在本地的节点，那么就叫 localhost
• PORT：Elasticsearch HTTP 服务所在的端口，默认为 9200 PATH API 路径（例如、_count 将返回集群中文档的数量），
• PATH：可以包含多个组件，例如 _cluster/stats 或者 _nodes/stats/jvm
• QUERY_STRING：一些可选的查询请求参数，例如？pretty 参数将使请求返回更加美观易读的 JSON 数据
• BODY：一个 JSON 格式的请求主体（如果请求需要的话）

ElasticSearch Rest API 分为两种：

• URI Search：在 URL 中使用查询参数
• Request Body Search：基于 JSON 格式的、更加完备的 DSL

URI Search 示例：

Request Body Search 示例：

索引 API

参考资料：Elasticsearch 官方之 cat 索引 API

创建索引

新建 Index，可以直接向 ES 服务器发出 PUT 请求。

语法格式：

1
2
3
4
5
6
7
8
9

PUT /my_index
{
    "settings": { ... any settings ... },
    "mappings": {
        "type_one": { ... any mappings ... },
        "type_two": { ... any mappings ... },
        ...
    }
}

示例：

1
2
3
4
5
6
7
8
9

PUT /user
{
  "settings": {
    "index": {
      "number_of_shards": 3,
      "number_of_replicas": 2
    }
  }
}

服务器返回一个 JSON 对象，里面的 acknowledged 字段表示操作成功。

1

{"acknowledged":true,"shards_acknowledged":true,"index":"user"}

如果你想禁止自动创建索引，可以通过在 config/elasticsearch.yml 的每个节点下添加下面的配置：

1

action.auto_create_index: false

删除索引

然后，我们可以通过发送 DELETE 请求，删除这个 Index。

1

DELETE /user

删除多个索引

1
2

DELETE /index_one,index_two
DELETE /index_*

查看索引

可以通过 GET 请求查看索引信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 查看索引相关信息
GET kibana_sample_data_ecommerce

# 查看索引的文档总数
GET kibana_sample_data_ecommerce/_count

# 查看前 10 条文档，了解文档格式
GET kibana_sample_data_ecommerce/_search

# _cat indices API
# 查看 indices
GET /_cat/indices/kibana*?v&s=index

# 查看状态为绿的索引
GET /_cat/indices?v&health=green

# 按照文档个数排序
GET /_cat/indices?v&s=docs.count:desc

# 查看具体的字段
GET /_cat/indices/kibana*?pri&v&h=health,index,pri,rep,docs.count,mt

# 查看索引占用的内存
GET /_cat/indices?v&h=i,tm&s=tm:desc

索引别名

ES 的索引别名就是给一个索引或者多个索引起的另一个名字，典型的应用场景是针对索引使用的平滑切换。

首先，创建索引 my_index，然后将别名 my_alias 指向它，示例如下：

1
2

PUT /my_index
PUT /my_index/_alias/my_alias

也可以通过如下形式：

1
2
3
4
5
6

POST /_aliases
{
  "actions": [
    { "add": { "index": "my_index", "alias": "my_alias" }}
  ]
}

也可以在一次请求中增加别名和移除别名混合使用：

1
2
3
4
5
6
7

POST /_aliases
{
  "actions": [
    { "remove": { "index": "my_index", "alias": "my_alias" }}
    { "add": { "index": "my_index_v2", "alias": "my_alias" }}
  ]
}

需要注意的是，如果别名与索引是一对一的，使用别名索引文档或者查询文档是可以的，但是如果别名和索引是一对多的，使用别名会发生错误，因为 ES 不知道把文档写入哪个索引中去或者从哪个索引中读取文档。

ES 索引别名有个典型的应用场景是平滑切换，更多细节可以查看 Elasticsearch（ES）索引零停机（无需重启）无缝平滑切换的方法。

打开/关闭索引

通过在 POST 中添加 _close 或 _open 可以打开、关闭索引。

打开索引

1
2
3
4

# 打开索引
POST kibana_sample_data_ecommerce/_open
# 关闭索引
POST kibana_sample_data_ecommerce/_close

文档

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

############Create Document############
#create document. 自动生成 _id
POST users/_doc
{
	"user" : "Mike",
    "post_date" : "2019-04-15T14:12:12",
    "message" : "trying out Kibana"
}

#create document. 指定 Id。如果 id 已经存在，报错
PUT users/_doc/1?op_type=create
{
    "user" : "Jack",
    "post_date" : "2019-05-15T14:12:12",
    "message" : "trying out Elasticsearch"
}

#create document. 指定 ID 如果已经存在，就报错
PUT users/_create/1
{
     "user" : "Jack",
    "post_date" : "2019-05-15T14:12:12",
    "message" : "trying out Elasticsearch"
}

### Get Document by ID
#Get the document by ID
GET users/_doc/1

###  Index & Update
#Update 指定 ID  （先删除，在写入）
GET users/_doc/1

PUT users/_doc/1
{
	"user" : "Mike"

}

#GET users/_doc/1
#在原文档上增加字段
POST users/_update/1/
{
    "doc":{
        "post_date" : "2019-05-15T14:12:12",
        "message" : "trying out Elasticsearch"
    }
}

### Delete by Id
# 删除文档
DELETE users/_doc/1

### Bulk 操作
#执行两次，查看每次的结果

#执行第 1 次
POST _bulk
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_id" : "2" } }
{ "create" : { "_index" : "test2", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_index" : "test"} }
{ "doc" : {"field2" : "value2"} }

#执行第 2 次
POST _bulk
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_id" : "2" } }
{ "create" : { "_index" : "test2", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_index" : "test"} }
{ "doc" : {"field2" : "value2"} }

### mget 操作
GET /_mget
{
    "docs" : [
        {
            "_index" : "test",
            "_id" : "1"
        },
        {
            "_index" : "test",
            "_id" : "2"
        }
    ]
}

#URI 中指定 index
GET /test/_mget
{
    "docs" : [
        {

            "_id" : "1"
        },
        {

            "_id" : "2"
        }
    ]
}

GET /_mget
{
    "docs" : [
        {
            "_index" : "test",
            "_id" : "1",
            "_source" : false
        },
        {
            "_index" : "test",
            "_id" : "2",
            "_source" : ["field3", "field4"]
        },
        {
            "_index" : "test",
            "_id" : "3",
            "_source" : {
                "include": ["user"],
                "exclude": ["user.location"]
            }
        }
    ]
}

### msearch 操作
POST kibana_sample_data_ecommerce/_msearch
{}
{"query" : {"match_all" : {}},"size":1}
{"index" : "kibana_sample_data_flights"}
{"query" : {"match_all" : {}},"size":2}

### 清除测试数据
#清除数据
DELETE users
DELETE test
DELETE test2

创建文档

指定 ID

语法格式：

1

PUT /_index/_type/_create/_id

示例：

1
2
3
4
5
6

PUT /user/_doc/_create/1
{
  "user": "张三",
  "title": "工程师",
  "desc": "数据库管理"
}

注意：指定 Id，如果 id 已经存在，则报错

自动生成 ID

新增记录的时候，也可以不指定 Id，这时要改成 POST 请求。

语法格式：

1

POST /_index/_type

示例：

1
2
3
4
5
6

POST /user/_doc
{
  "user": "张三",
  "title": "工程师",
  "desc": "超级管理员"
}

删除文档

语法格式：

1

DELETE /_index/_doc/_id

示例：

1

DELETE /user/_doc/1

更新文档

先删除，再写入

语法格式：

1

PUT /_index/_type/_id

示例：

1
2
3
4
5
6

PUT /user/_doc/1
{
  "user": "李四",
  "title": "工程师",
  "desc": "超级管理员"
}

在原文档上增加字段

语法格式：

1

POST /_index/_update/_id

示例：

1
2
3
4
5
6

POST /user/_update/1
{
    "doc":{
        "age" : "30"
    }
}

查询文档

指定 ID 查询

语法格式：

1

GET /_index/_type/_id

示例：

1

GET /user/_doc/1

结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

{
  "_index": "user",
  "_type": "_doc",
  "_id": "1",
  "_version": 1,
  "_seq_no": 536248,
  "_primary_term": 2,
  "found": true,
  "_source": {
    "user": "张三",
    "title": "工程师",
    "desc": "数据库管理"
  }
}

返回的数据中，found 字段表示查询成功，_source 字段返回原始记录。

如果 id 不正确，就查不到数据，found 字段就是 false

查询所有记录

使用 GET 方法，直接请求 /index/type/_search，就会返回所有记录。

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

$ curl 'localhost:9200/user/admin/_search?pretty'
{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 3,
    "successful" : 3,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 2,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "user",
        "_type" : "admin",
        "_id" : "WWuoDG8BHwECs7SiYn93",
        "_score" : 1.0,
        "_source" : {
          "user" : "李四",
          "title" : "工程师",
          "desc" : "系统管理"
        }
      },
      {
        "_index" : "user",
        "_type" : "admin",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "user" : "张三",
          "title" : "工程师",
          "desc" : "超级管理员"
        }
      }
    ]
  }
}

上面代码中，返回结果的 took字段表示该操作的耗时（单位为毫秒），timed_out字段表示是否超时，hits字段表示命中的记录，里面子字段的含义如下。

• total：返回记录数，本例是 2 条。
• max_score：最高的匹配程度，本例是1.0。
• hits：返回的记录组成的数组。

返回的记录中，每条记录都有一个_score字段，表示匹配的程序，默认是按照这个字段降序排列。

全文搜索

ES 的查询非常特别，使用自己的 查询语法，要求 GET 请求带有数据体。

1
2
3
4

$ curl -H 'Content-Type: application/json' 'localhost:9200/user/admin/_search?pretty'  -d '
{
"query" : { "match" : { "desc" : "管理" }}
}'

上面代码使用 Match 查询，指定的匹配条件是desc字段里面包含”软件”这个词。返回结果如下。

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

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 3,
    "successful" : 3,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 2,
    "max_score" : 0.38200712,
    "hits" : [
      {
        "_index" : "user",
        "_type" : "admin",
        "_id" : "WWuoDG8BHwECs7SiYn93",
        "_score" : 0.38200712,
        "_source" : {
          "user" : "李四",
          "title" : "工程师",
          "desc" : "系统管理"
        }
      },
      {
        "_index" : "user",
        "_type" : "admin",
        "_id" : "1",
        "_score" : 0.3487891,
        "_source" : {
          "user" : "张三",
          "title" : "工程师",
          "desc" : "超级管理员"
        }
      }
    ]
  }
}

Elastic 默认一次返回 10 条结果，可以通过size字段改变这个设置，还可以通过from字段，指定位移。

1
2
3
4
5
6

$ curl 'localhost:9200/user/admin/_search'  -d '
{
  "query" : { "match" : { "desc" : "管理" }},
  "from": 1,
  "size": 1
}'

上面代码指定，从位置 1 开始（默认是从位置 0 开始），只返回一条结果。

逻辑运算

如果有多个搜索关键字， Elastic 认为它们是or关系。

1
2
3
4

$ curl 'localhost:9200/user/admin/_search'  -d '
{
"query" : { "match" : { "desc" : "软件 系统" }}
}'

上面代码搜索的是软件 or 系统。

如果要执行多个关键词的and搜索，必须使用 布尔查询。

1
2
3
4
5
6
7
8
9
10
11

$ curl -H 'Content-Type: application/json' 'localhost:9200/user/admin/_search?pretty'  -d '
{
 "query": {
  "bool": {
   "must": [
    { "match": { "desc": "管理" } },
    { "match": { "desc": "超级" } }
   ]
  }
 }
}'

批量执行

支持在一次 API 调用中，对不同的索引进行操作

支持四种类型操作

• index
• create
• update
• delete

操作中单条操作失败，并不会影响其他操作。

返回结果包括了每一条操作执行的结果。

1
2
3
4
5
6
7
8

POST _bulk
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_id" : "2" } }
{ "create" : { "_index" : "test2", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_index" : "test"} }
{ "doc" : {"field2" : "value2"} }

说明：上面的示例如果执行多次，执行结果都不一样。

批量读取

读多个索引

1
2
3
4
5
6
7
8
9
10
11
12
13

GET /_mget
{
    "docs" : [
        {
            "_index" : "test",
            "_id" : "1"
        },
        {
            "_index" : "test",
            "_id" : "2"
        }
    ]
}

读一个索引

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

GET /test/_mget
{
    "docs" : [
        {

            "_id" : "1"
        },
        {

            "_id" : "2"
        }
    ]
}

GET /_mget
{
    "docs" : [
        {
            "_index" : "test",
            "_id" : "1",
            "_source" : false
        },
        {
            "_index" : "test",
            "_id" : "2",
            "_source" : ["field3", "field4"]
        },
        {
            "_index" : "test",
            "_id" : "3",
            "_source" : {
                "include": ["user"],
                "exclude": ["user.location"]
            }
        }
    ]
}

批量查询

1
2
3
4
5

POST kibana_sample_data_ecommerce/_msearch
{}
{"query" : {"match_all" : {}},"size":1}
{"index" : "kibana_sample_data_flights"}
{"query" : {"match_all" : {}},"size":2}

URI Search 查询语义

Elasticsearch URI Search 遵循 QueryString 查询语义，其形式如下：

1
2
3
4

GET /movies/_search?q=2012&df=title&sort=year:desc&from=0&size=10&timeout=1s
{
	"profile": true
}

• q 指定查询语句，使用 QueryString 语义
• df 默认字段，不指定时
• sort 排序：from 和 size 用于分页
• profile 可以查看查询时如何被执行的

1
2
3
4

GET /movies/_search?q=title:2012&sort=year:desc&from=0&size=10&timeout=1s
{
	"profile":"true"
}

Term 和 Phrase

Beautiful Mind 等效于 Beautiful OR Mind

“Beautiful Mind” 等效于 Beautiful AND Mind

1
2
3
4
5
6
7
8
9
10
11

# Term 查询
GET /movies/_search?q=title:Beautiful Mind
{
	"profile":"true"
}

# 使用引号，Phrase 查询
GET /movies/_search?q=title:"Beautiful Mind"
{
	"profile":"true"
}

分组与引号

title:(Beautiful AND Mind)

title=”Beautiful Mind”

AND、OR、NOT 或者 &&、||、!

注意：AND、OR、NOT 必须大写

1
2
3
4
5
6
7
8
9
10

# 布尔操作符
GET /movies/_search?q=title:(Beautiful AND Mind)
{
	"profile":"true"
}

GET /movies/_search?q=title:(Beautiful NOT Mind)
{
	"profile":"true"
}

范围查询

• [] 表示闭区间
• {} 表示开区间

示例：

1
2
3
4
5
6
7
8
9
10

# 范围查询 , 区间写法
GET /movies/_search?q=title:beautiful AND year:{2010 TO 2018%7D
{
	"profile":"true"
}

GET /movies/_search?q=title:beautiful AND year:[* TO 2018]
{
	"profile":"true"
}

算数符号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 2010 年以后的记录
GET /movies/_search?q=year:>2010
{
	"profile":"true"
}

# 2010 年到 2018 年的记录
GET /movies/_search?q=year:(>2010 && <=2018)
{
	"profile":"true"
}

# 2010 年到 2018 年的记录
GET /movies/_search?q=year:(+>2010 +<=2018)
{
	"profile":"true"
}

通配符查询

• ? 代表 1 个字符
• * 代表 0 或多个字符

示例：

1
2
3
4
5
6
7
8
9

GET /movies/_search?q=title:mi?d
{
	"profile":"true"
}

GET /movies/_search?q=title:b*
{
	"profile":"true"
}

正则表达式

title:[bt]oy

模糊匹配与近似查询

示例：

1
2
3
4
5
6
7
8
9
10
11

# 相似度在 1 个字符以内
GET /movies/_search?q=title:beautifl~1
{
	"profile":"true"
}

# 相似度在 2 个字符以内
GET /movies/_search?q=title:"Lord Rings"~2
{
	"profile":"true"
}

Request Body & DSL

Elasticsearch 除了 URI Search 查询方式，还支持将查询语句通过 Http Request Body 发起查询。

1
2
3
4
5
6
7

GET /kibana_sample_data_ecommerce/_search?ignore_unavailable=true
{
	"profile":"true",
	"query": {
	  "match_all": {}
	}
}

分页

1
2
3
4
5
6
7
8
9

GET /kibana_sample_data_ecommerce/_search?ignore_unavailable=true
{
  "profile": "true",
  "from": 0,
  "size": 10,
  "query": {
    "match_all": {}
  }
}

排序

最好在数字型或日期型字段上排序

因为对于多值类型或分析过的字段排序，系统会选一个值，无法得知该值

1
2
3
4
5
6
7
8
9
10
11
12
13
14

GET /kibana_sample_data_ecommerce/_search?ignore_unavailable=true
{
  "profile": "true",
  "sort": [
    {
      "order_date": "desc"
    }
  ],
  "from": 1,
  "size": 10,
  "query": {
    "match_all": {}
  }
}

_source 过滤

如果 _source 没有存储，那就只返回匹配的文档的元数据

_source 支持使用通配符，如：_source["name*", "desc*"]

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13

GET /kibana_sample_data_ecommerce/_search?ignore_unavailable=true
{
  "profile": "true",
  "_source": [
    "order_date",
    "category.keyword"
  ],
  "from": 1,
  "size": 10,
  "query": {
    "match_all": {}
  }
}

脚本字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

GET /kibana_sample_data_ecommerce/_search?ignore_unavailable=true
{
  "profile": "true",
  "script_fields": {
    "new_field": {
      "script": {
        "lang": "painless",
        "source":"doc['order_date'].value+' hello'"
      }
    }
  },
  "from": 1,
  "size": 10,
  "query": {
    "match_all": {}
  }
}

使用查询表达式 - Match

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

POST movies/_search
{
  "query": {
    "match": {
      "title": "last christmas"
    }
  }
}

POST movies/_search
{
  "query": {
    "match": {
      "title": {
        "query": "last christmas",
        "operator": "and"
      }
    }
  }
}

短语搜索 - Match Phrase

1
2
3
4
5
6
7
8
9
10
11

POST movies/_search
{
  "query": {
    "match_phrase": {
      "title":{
        "query": "last christmas"

      }
    }
  }
}

集群 API

Elasticsearch 官方之 Cluster API

一些集群级别的 API 可能会在节点的子集上运行，这些节点可以用节点过滤器指定。例如，任务管理、节点统计和节点信息 API 都可以报告来自一组过滤节点而不是所有节点的结果。

节点过滤器以逗号分隔的单个过滤器列表的形式编写，每个过滤器从所选子集中添加或删除节点。每个过滤器可以是以下之一：

• _all：将所有节点添加到子集
• _local：将本地节点添加到子集
• _master：将当前主节点添加到子集
• 根据节点 ID 或节点名将匹配节点添加到子集
• 根据 IP 地址或主机名将匹配节点添加到子集
• 使用通配符，将节点名、地址名或主机名匹配的节点添加到子集
• master:true, data:true, ingest:true, voting_only:true, ml:true 或 coordinating_only:true, 分别意味着将所有主节点、所有数据节点、所有摄取节点、所有仅投票节点、所有机器学习节点和所有协调节点添加到子集中。
• master:false, data:false, ingest:false, voting_only:true, ml:false 或 coordinating_only:false, 分别意味着将所有主节点、所有数据节点、所有摄取节点、所有仅投票节点、所有机器学习节点和所有协调节点排除在子集外。
• 配对模式，使用 * 通配符，格式为 attrname:attrvalue，将所有具有自定义节点属性的节点添加到子集中，其名称和值与相应的模式匹配。自定义节点属性是通过 node.attr.attrname: attrvalue 形式在配置文件中设置的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 如果没有给出过滤器，默认是查询所有节点
GET /_nodes
# 查询所有节点
GET /_nodes/_all
# 查询本地节点
GET /_nodes/_local
# 查询主节点
GET /_nodes/_master
# 根据名称查询节点（支持通配符）
GET /_nodes/node_name_goes_here
GET /_nodes/node_name_goes_*
# 根据地址查询节点（支持通配符）
GET /_nodes/10.0.0.3,10.0.0.4
GET /_nodes/10.0.0.*
# 根据规则查询节点
GET /_nodes/_all,master:false
GET /_nodes/data:true,ingest:true
GET /_nodes/coordinating_only:true
GET /_nodes/master:true,voting_only:false
# 根据自定义属性查询节点（如：查询配置文件中含 node.attr.rack:2 属性的节点）
GET /_nodes/rack:2
GET /_nodes/ra*:2
GET /_nodes/ra*:2*

集群健康 API

1
2
3
4

GET /_cluster/health
GET /_cluster/health?level=shards
GET /_cluster/health/kibana_sample_data_ecommerce,kibana_sample_data_flights
GET /_cluster/health/kibana_sample_data_flights?level=shards

集群状态 API

集群状态 API 返回表示整个集群状态的元数据。

1

GET /_cluster/state

节点 API

Elasticsearch 官方之 cat Nodes API——返回有关集群节点的信息。

1
2
3
4

# 查看默认的字段
GET /_cat/nodes?v=true
# 查看指定的字段
GET /_cat/nodes?v=true&h=id,ip,port,v,m

分片 API

Elasticsearch 官方之 cat Shards API——shards 命令是哪些节点包含哪些分片的详细视图。它会告诉你它是主还是副本、文档数量、它在磁盘上占用的字节数以及它所在的节点。

1
2
3
4
5
6

# 查看默认的字段
GET /_cat/shards
# 根据名称查询分片（支持通配符）
GET /_cat/shards/my-index-*
# 查看指定的字段
GET /_cat/shards?h=index,shard,prirep,state,unassigned.reason

监控 API

Elasticsearch 中集群相关的健康、统计等相关的信息都是围绕着 cat API 进行的。

通过 GET 请求发送 cat，下面列出了所有可用的 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

GET /_cat

=^.^=
/_cat/allocation
/_cat/shards
/_cat/shards/{index}
/_cat/master
/_cat/nodes
/_cat/tasks
/_cat/indices
/_cat/indices/{index}
/_cat/segments
/_cat/segments/{index}
/_cat/count
/_cat/count/{index}
/_cat/recovery
/_cat/recovery/{index}
/_cat/health
/_cat/pending_tasks
/_cat/aliases
/_cat/aliases/{alias}
/_cat/thread_pool
/_cat/thread_pool/{thread_pools}
/_cat/plugins
/_cat/fielddata
/_cat/fielddata/{fields}
/_cat/nodeattrs
/_cat/repositories
/_cat/snapshots/{repository}
/_cat/templates

参考资料

• Elasticsearch 官方文档
• https://docs.spring.io/spring-data/elasticsearch/docs/current/reference/html/#elasticsearch.clients

Elasticsearch CRUD

::: info 概述

CRUD 由英文单词 Create, Read, Update, Delete 的首字母组成，即增删改查。

本文通过介绍基本的 Elasticsearch CRUD 方法，向读者呈现如何访问 Elasticsearch 数据。
:::

Elasticsearch 简介

::: info 概述

Elasticsearch 是一个基于 Lucene 的搜索和数据分析工具，它提供了一个分布式服务。Elasticsearch 是遵从 Apache 开源条款的一款开源产品，是当前主流的企业级搜索引擎。

本文简单介绍了 Elasticsearch 的功能、特性、简史、概念，可以让读者在短时间内对于 Elasticsearch 有一个初步的认识。

:::

Elasticsearch 运维

Elasticsearch 是一个分布式、RESTful 风格的搜索和数据分析引擎，能够解决不断涌现出的各种用例。 作为 Elastic Stack 的核心，它集中存储您的数据，帮助您发现意料之中以及意料之外的情况。

Elasticsearch 安装

Elasticsearch 官方下载安装说明

（1）下载解压

访问 官方下载地址 ，选择需要的版本，下载解压到本地。

（2）运行

运行 bin/elasticsearch (Windows 系统上运行 bin\elasticsearch.bat )

（3）访问

执行 curl http://localhost:9200/ 测试服务是否启动

Elasticsearch 集群规划

ElasticSearch 集群需要根据业务实际情况去合理规划。

需要考虑的问题点：

• 集群部署几个节点？
• 有多少个索引？
• 每个索引有多大数据量？
• 每个索引有多少个分片？

一个参考规划：

• 3 台机器，每台机器是 6 核 64G 的。
• 我们 es 集群的日增量数据大概是 2000 万条，每天日增量数据大概是 500MB，每月增量数据大概是 6 亿，15G。目前系统已经运行了几个月，现在 es 集群里数据总量大概是 100G 左右。
• 目前线上有 5 个索引（这个结合你们自己业务来，看看自己有哪些数据可以放 es 的），每个索引的数据量大概是 20G，所以这个数据量之内，我们每个索引分配的是 8 个 shard，比默认的 5 个 shard 多了 3 个 shard。

Elasticsearch 配置

ES 的默认配置文件为 config/elasticsearch.yml

基本配置说明如下：

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

cluster.name: elasticsearch
#配置es的集群名称，默认是elasticsearch，es会自动发现在同一网段下的es，如果在同一网段下有多个集群，就可以用这个属性来区分不同的集群。
node.name: 'Franz Kafka'
#节点名，默认随机指定一个name列表中名字，该列表在es的jar包中config文件夹里name.txt文件中，其中有很多作者添加的有趣名字。
node.master: true
#指定该节点是否有资格被选举成为node，默认是true，es是默认集群中的第一台机器为master，如果这台机挂了就会重新选举master。
node.data: true
#指定该节点是否存储索引数据，默认为true。
index.number_of_shards: 5
#设置默认索引分片个数，默认为5片。
index.number_of_replicas: 1
#设置默认索引副本个数，默认为1个副本。
path.conf: /path/to/conf
#设置配置文件的存储路径，默认是es根目录下的config文件夹。
path.data: /path/to/data
#设置索引数据的存储路径，默认是es根目录下的data文件夹，可以设置多个存储路径，用逗号隔开，例：
#path.data: /path/to/data1,/path/to/data2
path.work: /path/to/work
#设置临时文件的存储路径，默认是es根目录下的work文件夹。
path.logs: /path/to/logs
#设置日志文件的存储路径，默认是es根目录下的logs文件夹
path.plugins: /path/to/plugins
#设置插件的存放路径，默认是es根目录下的plugins文件夹
bootstrap.mlockall: true
#设置为true来锁住内存。因为当jvm开始swapping时es的效率会降低，所以要保证它不swap，可以把#ES_MIN_MEM和ES_MAX_MEM两个环境变量设置成同一个值，并且保证机器有足够的内存分配给es。同时也要#允许elasticsearch的进程可以锁住内存，linux下可以通过`ulimit -l unlimited`命令。
network.bind_host: 192.168.0.1
#设置绑定的ip地址，可以是ipv4或ipv6的，默认为0.0.0.0。
network.publish_host: 192.168.0.1
#设置其它节点和该节点交互的ip地址，如果不设置它会自动判断，值必须是个真实的ip地址。
network.host: 192.168.0.1
#这个参数是用来同时设置bind_host和publish_host上面两个参数。
transport.tcp.port: 9300
#设置节点间交互的tcp端口，默认是9300。
transport.tcp.compress: true
#设置是否压缩tcp传输时的数据，默认为false，不压缩。
http.port: 9200
#设置对外服务的http端口，默认为9200。
http.max_content_length: 100mb
#设置内容的最大容量，默认100mb
http.enabled: false
#是否使用http协议对外提供服务，默认为true，开启。
gateway.type: local
#gateway的类型，默认为local即为本地文件系统，可以设置为本地文件系统，分布式文件系统，hadoop的#HDFS，和amazon的s3服务器，其它文件系统的设置方法下次再详细说。
gateway.recover_after_nodes: 1
#设置集群中N个节点启动时进行数据恢复，默认为1。
gateway.recover_after_time: 5m
#设置初始化数据恢复进程的超时时间，默认是5分钟。
gateway.expected_nodes: 2
#设置这个集群中节点的数量，默认为2，一旦这N个节点启动，就会立即进行数据恢复。
cluster.routing.allocation.node_initial_primaries_recoveries: 4
#初始化数据恢复时，并发恢复线程的个数，默认为4。
cluster.routing.allocation.node_concurrent_recoveries: 2
#添加删除节点或负载均衡时并发恢复线程的个数，默认为4。
indices.recovery.max_size_per_sec: 0
#设置数据恢复时限制的带宽，如入100mb，默认为0，即无限制。
indices.recovery.concurrent_streams: 5
#设置这个参数来限制从其它分片恢复数据时最大同时打开并发流的个数，默认为5。
discovery.zen.minimum_master_nodes: 1
#设置这个参数来保证集群中的节点可以知道其它N个有master资格的节点。默认为1，对于大的集群来说，可以设置大一点的值（2-4）
discovery.zen.ping.timeout: 3s
#设置集群中自动发现其它节点时ping连接超时时间，默认为3秒，对于比较差的网络环境可以高点的值来防止自动发现时出错。
discovery.zen.ping.multicast.enabled: false
#设置是否打开多播发现节点，默认是true。
discovery.zen.ping.unicast.hosts: ['host1', 'host2:port', 'host3[portX-portY]']
#设置集群中master节点的初始列表，可以通过这些节点来自动发现新加入集群的节点。

Elasticsearch FAQ

elasticsearch 不允许以 root 权限来运行

问题：在 Linux 环境中，elasticsearch 不允许以 root 权限来运行。

如果以 root 身份运行 elasticsearch，会提示这样的错误：

1

can not run elasticsearch as root

解决方法：使用非 root 权限账号运行 elasticsearch

1
2
3
4
5
6
7
8

# 创建用户组
groupadd elk
# 创建新用户，-g elk 设置其用户组为 elk，-p elk 设置其密码为 elk
useradd elk -g elk -p elk
# 更改 /opt 文件夹及内部文件的所属用户及组为 elk:elk
chown -R elk:elk /opt # 假设你的 elasticsearch 安装在 opt 目录下
# 切换账号
su elk

vm.max_map_count 不低于 262144

问题：vm.max_map_count 表示虚拟内存大小，它是一个内核参数。elasticsearch 默认要求 vm.max_map_count 不低于 262144。

1

max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

解决方法：

你可以执行以下命令，设置 vm.max_map_count ，但是重启后又会恢复为原值。

1

sysctl -w vm.max_map_count=262144

持久性的做法是在 /etc/sysctl.conf 文件中修改 vm.max_map_count 参数：

1
2

echo "vm.max_map_count=262144" > /etc/sysctl.conf
sysctl -p

注意

如果运行环境为 docker 容器，可能会限制执行 sysctl 来修改内核参数。

这种情况下，你只能选择直接修改宿主机上的参数了。

nofile 不低于 65536

问题： nofile 表示进程允许打开的最大文件数。elasticsearch 进程要求可以打开的最大文件数不低于 65536。

1

max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]

解决方法：

在 /etc/security/limits.conf 文件中修改 nofile 参数：

1
2

echo "* soft nofile 65536" > /etc/security/limits.conf
echo "* hard nofile 131072" > /etc/security/limits.conf

nproc 不低于 2048

问题： nproc 表示最大线程数。elasticsearch 要求最大线程数不低于 2048。

1

max number of threads [1024] for user [user] is too low, increase to at least [2048]

解决方法：

在 /etc/security/limits.conf 文件中修改 nproc 参数：

1
2