在今天的文章中，我们将讲述如何运用 Elasticsearch 的 ingest 节点来对数据进行结构化并对数据进行处理。

数据集

在我们的实际数据采集中，数据可能来自不同的来源，并且以不同的形式展展现：

这些数据可以是一种很结构化的数据被摄，比如数据库中的数据，或者就是一直最原始的非结构化的数据，比如日志。对于一些非结构化的数据，我们该如何把它们结构化，并使用 Elasticsearch 进行分析呢？

结构化数据

就如上面的数据展示的那样。在很多的情况下，数据在摄入的时候是一种非结构化的形式来呈现的。这个数据通常有一个叫做 message 的字段。为了能达到结构化的目的，我们们需要 parse 及 transform 这个 message 字段，并把这个 message 变为我们所需要的字段，从而达到结构化的目的。让我们看一个例子。假如我们有如下的信息：

{"message": "2019-09-29T00:39:02.9122 [Debug] MyApp stopped"
}

显然上面的信息是一个非结构化的信息。它含有唯一的一个字段 message。我们希望通过一些方法把它变成为：

{"@timestamp": "2019-09-29T00:39:02.9122","loglevel": "Debug","status": "MyApp stopped"
}

显然上面的数据是一个结构化的文档。它更便于我们对数据进行分析。比如我们对数据进行聚合或在 Kibana 中进行展示。

我们接下来看一下一个典型的 Elastic Stack 的架构图：

在上面，我们可以看到有三个地方我们可以对数据进行处理：

我们可以使用 Beats processors, Logstash 和 Ingest node 来对我们的数据进行处理。如果大家还对使用 Logstash 或者是 Ingest Node 没法做选择的话，请参阅我之前的文章 “我应该使用 Logstash 或是 Elasticsearch ingest 节点?”。

如果你的日志数据不是一个已有的格式，比如 apache, nginx，那么你需要建立自己的 pipeline 来对这些日志进行处理。在今天的文章里，我们将介绍如何使用 Elasticsearch 的 ingest processors 来对我们的非结构化数据进行处理，从而把它们变为结构化的数据：

split
dissect
kv
grok
...

Ingest pipelines

一个 Elasticsearch pipeline 是一组 processors:

让我们在数据建立索引之前做预处理
每一个 processor 可以修改经过它的文档
processor 的处理是在 Elasticsearch 新的 ingest node 里进行的

定义一个 Elasticsearch 的 ingest pipeline

我们可以使用 Ingest API 来定义 pipelines:

我们可以使用 _simulate 终点来进行测试：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " "}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK"}}]
}

上面的运行的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"message" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK"]},"_ingest" : {"timestamp" : "2020-04-27T08:40:43.059569Z"}}}]
}

我们看到在上面的 split proocessor 中它把一个非结构化的 message 变成了一个结果话的数据。message 现在是一个数组，那么我们该如何引用这个数组里的数据呢？

我们接着修改 pipeline：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " "}},{"set": {"field": "timestamp","value": "{{message.0}}"}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK"}}]
}

在上面我们使用了 {{message.0}} 来访问数组里的第一个数据。上面的命令运行的结果为：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"message" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK"],"timestamp" : "2019-09-29T00:39:02.912Z"},"_ingest" : {"timestamp" : "2020-12-09T02:08:25.004644Z"}}}]
}

我们可以看到一个叫做 timestamp 的字段。

在实际的使用中，我们甚至可以使用 target_field 来重新被 split 后的字段名称：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " ","target_field": "new"}},{"set": {"field": "timestamp","value": "{{message.0}}"}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000"}}]
}

上面运行的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"new" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK","2000"],"message" : "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000","timestamp" : ""},"_ingest" : {"timestamp" : "2020-12-09T02:13:43.697296Z"}}}]
}

我们可以看到一个叫做 new 的字段代替之前的 message。由于我们增加了一个新的文字 “2000”，在我们的 new 字段输出中，可以看到一个新增加的字符串 “2000”。假如我们想把这个字段转换为整数，那么我们可以使用如下的办法：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " ","target_field": "new"}},{"set": {"field": "timestamp","value": "{{message.0}}"}},{"convert": {"field": "new.3","type": "integer"}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000"}}]
}

在上面，我们使用 new.3 来表想要转换的字段。上面的输出结果为：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"new" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK",2000],"message" : "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000","timestamp" : ""},"_ingest" : {"timestamp" : "2020-12-09T02:16:30.144772Z"}}}]
}

从上面我们可以看出来 “2000” 变成了 2000。

在实际使用 ingest pipeline 时，我们还会经常使用 conditonal 来对一些事件进行特殊的处理，比如我们在使用 set processor 时，可以针对一些条件进行特别的设置：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " ","target_field": "new"}},{"set": {"field": "timestamp","value": "{{message.0}}"}},{"convert": {"field": "new.3","type": "integer"}},{"set": {"if": "ctx.new.2 ==\"STATUS_OK\"","field": "good","value": true}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000"}}]
}

在上面，当我们的 ctx.new.2 的值是 "STATUS_OK" 时，我们可以增加一个字段叫做 good，并设置其值为 true。上面运行的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"new" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK",2000],"message" : "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000","good" : true,"timestamp" : ""},"_ingest" : {"timestamp" : "2021-07-04T02:13:46.535685Z"}}}]
}

上面的命令的格式也可以写作为：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " ","target_field": "new"}},{"set": {"field": "timestamp","value": "{{message.0}}"}},{"convert": {"field": "new.3","type": "integer"}},{"set": {"if": "ctx['new'][2] ==\"STATUS_OK\"","field": "good","value": true}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK 2000"}}]
}

如何使用 Pipeline

一旦你定义好一个 pipeline，如果你是使用 Filebeat 接入到 Elasticsearch 导入数据，那么你可以在 filebeat 的配置文件中这样使用这个 pipeline：

output.elasticsearch:hosts: ["http://localhost:9200"]pipeline: my_pipeline

你也可以直接为你的 Elasticsearch index 定义一个默认的 pipeline：

PUT my_index
{"settings": {"default_pipeline": "my_pipeline"}
}

这样当我们的数据导入到 my_index 里去的时候，my_pipeline 将会被自动调用。

例子

Dissect

我们下面来看一个更为复杂一点的例子。你需要同时使用 split 及 kv processor 来结构化这个消息：

正如我们上面显示的那样，我们想提取上面用红色标识的部分，但是我们并不需要信息中的中括号 “[” 及 “]”。我可以使用 dissect processor:

POST _ingest/pipeline/_simulate
{"pipeline": {"description": "Example using dissect processor","processors": [{"dissect": {"field": "message","pattern": "%{@timestamp} [%{loglevel}] %{status}"}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z [Debug] MyApp stopped"}}]
}

上面显示的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"@timestamp" : "2019-09-29T00:39:02.912Z","loglevel" : "Debug","message" : "2019-09-29T00:39:02.912Z [Debug] MyApp stopped","status" : "MyApp stopped"},"_ingest" : {"timestamp" : "2020-04-27T09:10:33.720665Z"}}}]
}

我们接下来显示一个 key-value 对的信息：

{"message": "2019009-29T00:39:02.912Z host=AppServer status=STATUS_OK"
}

我们同样可以使用 dissect processor 来处理：

POST _ingest/pipeline/_simulate
{"pipeline": {"description": "Example using dissect processor key-value","processors": [{"dissect": {"field": "message","pattern": "%{@timestamp} %{*field1}=%{&field1} %{*field2}=%{&field2}"}}]},"docs": [{"_source": {"message": "2019009-29T00:39:02.912Z host=AppServer status=STATUS_OK"}}]
}

在上面，*及 & 是参考键修饰符，它们用来改变 dissect 的行为。上面的结果显示：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"@timestamp" : "2019009-29T00:39:02.912Z","host" : "AppServer","message" : "2019009-29T00:39:02.912Z host=AppServer status=STATUS_OK","status" : "STATUS_OK"},"_ingest" : {"timestamp" : "2020-04-27T14:04:38.639127Z"}}}]
}

对于许多新的开发者来说，有时他们对 dissect 和 grok 的区别不是很理解。从表面上看，dissect 和 grok 有很多重叠的地方，但是 dissect 的执行速度远远高于 grok，所以在实际的使用中，尽量使用 dissect 来完成。但是在实际的使用中，有些情况下，我们还必须使用 grok 来完成。我们在一下的 grok 部分讲到。

Script processor

尽管现有的很多的 processor 都能给我们带来很大的方便，但是在实际的应用中，有很多的能够并不在我们的 Logstash 或 Elasticsearch 预设的功能之列。一种办法就是写自己的插件，但是这可能是一件巨大的任务。我们可以写一个脚本来完成这个工作。通常这个是由Elasticsearch的Painless脚本来完成的。如果你想了解更多的Painless的知识，你可以在 “Elastic：菜鸟上手指南” 找到几篇这个语言的介绍文章。

有两种方法可以允许 Painless script：inline 或者 stored。

Inline scripts

在下面的例子中它展示的是一个 inline 的脚本，用来更新一个叫做 new_field 的字段：

PUT /_ingest/pipeline/my_script_pipeline
{"processors": [{"script": {"source": "ctx['new_field'] = params.value","params": {"value": "Hello world"}}}]
}

在上面，我们使用 params 来把参数传入。这样做的好处是 source 的代码一直是没有变化的，这样它只会被编译一次。如果 source 的代码随着调用的不同而改变，那么它将会被每次编译从而造成浪费。

Stored scripts

Scripts也可以保存于 Cluster 的状态中，并且在以后引用 script 的 ID 来调用：

PUT _scripts/my_script
{"script": {"lang": "painless","source": "ctx['new_field'] = params.value"}
}PUT /_ingest/pipeline/my_script_pipeline
{"processors": [{"script": {"id": "my_script","params": {"value": "Hello world!"}}}]
}

上面的两个命令将实现和之前一样的功能。当我们在 ingest node 使用场景的时候，我们访问文档的字段时，使用 cxt['new_field']。我们也可以访问它的元字段，比如 cxt['_id'] = ctx['my_field']。

我们先来做几个练习：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"script": {"lang": "painless","source": "ctx['new_value'] = ctx['current_value'] + 1"}}]},"docs": [{"_source": {"current_value": 2}}]
}

上面的脚本运行时会生产一个新的叫做 new_value 的字段，并且它的值将会是由 curent_value 字段的值加上1。运行上面的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"new_value" : 3,"current_value" : 2},"_ingest" : {"timestamp" : "2020-04-27T14:49:35.775395Z"}}}]
}

我们接下来一个例子就是来创建一个 stored script:

PUT _scripts/my_script
{"script": {"lang": "painless","source": "ctx['new_value'] = ctx['current_value'] + params.value"}
}PUT /_ingest/pipeline/my_script_pipeline
{"processors": [{"script": {"id": "my_script","params": {"value": 1}}}]
}

上面的这个语句和之前的那个实现的是同一个功能。我们先执行上面的两个命令。为了能测试上面的 pipeline 是否工作，我们尝试创建两个文档：

POST test_docs/_doc
{"current_value": 34
}POST test_docs/_doc
{"current_value": 80
}

然后，我们运行如下的命令：

POST test_docs/_update_by_query?pipeline=my_script_pipeline
{"query": {"range": {"current_value": {"gt": 30}}}
}

在上面，我们通过使用 _update_by_query 结合 pipepline 一起来更新我们的文档。我们只针对 current_value 大于30的文档才起作用。运行完后：

{"took" : 25,"timed_out" : false,"total" : 2,"updated" : 2,"deleted" : 0,"batches" : 1,"version_conflicts" : 0,"noops" : 0,"retries" : {"bulk" : 0,"search" : 0},"throttled_millis" : 0,"requests_per_second" : -1.0,"throttled_until_millis" : 0,"failures" : [ ]
}

它显示已经更新两个文档了。我们使用如下的语句来进行查看：

GET test_docs/_search

显示的结果：

{"took" : 0,"timed_out" : false,"_shards" : {"total" : 1,"successful" : 1,"skipped" : 0,"failed" : 0},"hits" : {"total" : {"value" : 2,"relation" : "eq"},"max_score" : 1.0,"hits" : [{"_index" : "test_docs","_type" : "_doc","_id" : "EIEnvHEBQHMgxFmxZyBq","_score" : 1.0,"_source" : {"new_value" : 35,"current_value" : 34}},{"_index" : "test_docs","_type" : "_doc","_id" : "D4EnvHEBQHMgxFmxXyBp","_score" : 1.0,"_source" : {"new_value" : 81,"current_value" : 80}}]}
}

从上面我们可以看出来 new_value 字段的值是 current_value 字段的值加上1。

我们再接着看如下的例子：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"split": {"field": "message","separator": " ", "target_field": "split_message"}},{"set": {"field": "environment","value": "prod"}},{"set": {"field": "@timestamp","value": "{{split_message.0}}"}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK"}}]
}

在上面第一个 split processor，我们把 message 按照" "来进行拆分，并同时把结果赋予给字段 split_message。它其实是一个数组。接着我们通过 set processor添加一个叫做 environment 的字段，并赋予值 prod。再接着我们把 split_message 数组里的第一个值拿出来赋予给 @timestamp 字段。这是一个添加的字段。运行的结果如下：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"environment" : "prod","@timestamp" : "2019-09-29T00:39:02.912Z","message" : "2019-09-29T00:39:02.912Z AppServer1 STATUS_OK","split_message" : ["2019-09-29T00:39:02.912Z","AppServer1","STATUS_OK"]},"_ingest" : {"timestamp" : "2020-04-27T15:35:00.922037Z"}}}]
}

Grok processor

Grok processor 提供了一种正则匹配的方式让我们把 pattern 和 message 进行匹配，从而提前出 message 里的结构化数据。相比较 Dissect 而言，Grok 的相率并不高。这是我们需要注意的。那么为什么我们还是需要使用 Grok呢？我们首先来看一下如下的一个例子：

157.97.192.70 2019 09 29 00:39:02.912 AppServer1 Process 11111 Init
157.97.192.70 2019 09 29 00:39:06.554 AppServer1 22222 Stopped 3.642

在上面的两个日志中，我们发现如果使用 Dissect processor，还是无能为力，这是因为 process id 在两个不同的日志里出现的位置并不相同。但是我们可以使用 Grok 来完美地解决这个问题。

我们可以在 Kibana 中打入如下的命令来查询现有的预设的 grok pattern:

GET /_ingest/processor/grok

我们可以看到有超过 300 多个的预设的 grok patern 供我们使用：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"grok": {"field": "message","patterns": ["%{TIMESTAMP_ISO8601:@timestamp} %{IP:client} \\[%{WORD:status}\\] %{NUMBER:duration}"]}}]},"docs": [{"_source": {"message": "2019-09-29T00:39:02.912Z 55.3.241.1 [OK] 0.043"}}]
}

上面的返回结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"duration" : "0.043","@timestamp" : "2019-09-29T00:39:02.912Z","client" : "55.3.241.1","message" : "2019-09-29T00:39:02.912Z 55.3.241.1 [OK] 0.043","status" : "OK"},"_ingest" : {"timestamp" : "2020-04-28T00:16:52.155688Z"}}}]
}

Grok processro 也对多行的事件也可以处理的很好。比如：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"grok": {"field": "text","patterns": ["%{GREEDYMULTILINE:allMyData}"],"pattern_definitions": {"GREEDYMULTILINE": "(.|\n)*"}}}]},"docs": [{"_source": {"text": "This is a text \n secondline"}}]
}

上面运行的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"text" : """This is a text secondline""","allMyData" : """This is a text secondline"""},"_ingest" : {"timestamp" : "2020-04-28T00:31:38.913929Z"}}}]
}

在上面我们可以看到 allMydata 把多行的数据都提前到同一个字段。在上面如果我们只用其中的一种 pattern_definitions，比如 .*：

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"grok": {"field": "text","patterns": ["%{GREEDYMULTILINE:allMyData}"],"pattern_definitions": {"GREEDYMULTILINE": ".*"}}}]},"docs": [{"_source": {"text": "This is a text \n secondline"}}]
}

那么我们可以看到：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"text" : """This is a text secondline""","allMyData" : "This is a text "},"_ingest" : {"timestamp" : "2020-04-28T00:35:59.67759Z"}}}]
}

也就是它只提前了第一行。

Date processor

POST /_ingest/pipeline/_simulate
{"pipeline": {"processors": [{"date": {"field": "date","formats": ["MM/dd/yyyy HH:mm","dd-MM-yyyy HH:mm:ssz"]}}]},"docs": [{"_source": {"date": "03/25/2019 03:39"}},{"_source": {"date": "25-03-2019 03:39:00+01:00"}}]
}

在上面我们定义了两种时间的格式，如果其中的一个有匹配，那么时间将会被正确地解析，同时被自动赋予给 @timestamp 字段。这个和 Logstash 的 date processor 是一样的。上面运行的结果是：

{"docs" : [{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"date" : "03/25/2019 03:39","@timestamp" : "2019-03-25T03:39:00.000Z"},"_ingest" : {"timestamp" : "2020-04-28T00:24:24.802381Z"}}},{"doc" : {"_index" : "_index","_type" : "_doc","_id" : "_id","_source" : {"date" : "25-03-2019 03:39:00+01:00","@timestamp" : "2019-03-25T02:39:00.000Z"},"_ingest" : {"timestamp" : "2020-04-28T00:24:24.802396Z"}}}]
}

Elasticsearch：Elastic可观测性 - 运用 pipeline 使数据结构化相关推荐

Elasticsearch：创建 Ingest pipeline
在 Elasticsearch 针对数据进行分析之前,我们必须针对数据进行摄入.在摄入的过程中,我们需要对数据进行加工,这其中包括非结构化数据转换为结构化数据,数据的转换,丰富,删除,添加新的字段等等 ...
百度智慧医疗总经理黄艳：基层筛查、临床辅助决策和医疗数据结构化的阶段性进展...
雷锋网消息,4月1日,2019国际医学人工智能论坛暨ITU与WHO健康医疗人工智能焦点组会议在上海举办.在医学人工智能新技术创新环节,百度AI创新业务部高级总监.智慧医疗总经理黄艳发表了演讲. 百度灵 ...
直播 | 达观数据高级技术专家杨慧宇：金融数据结构化实践
「PW Live」是由 PaperWeekly 和 biendata 共同发起的学术直播间,旨在帮助更多的青年学者宣传其最新科研成果.我们一直认为,单向地输出知识并不是一个最好的方式,而有效地反馈和交 ...
Logstash+elasticsearch+elastic+nignx
注:本系统使用的是Logstash+elasticsearch+elastic+nignx 进行日志分析.展示 1环境版本:... 2 1.1主机:... 2 1.2前提:... 2 2 Logsta ...
项目练习（二）—微博数据结构化
1.ETL概念 ETL,是英文 Extract-Transform-Load 的缩写,用来描述将数据从来源端经过抽取(extract).交互转换(transform).加载(load)至 ...
使用twig来组装数据使数据结构可视化
业务场景: 第三方平台实现微信图文,在页面上用ajax加载更多图文时需要组装大量的JSON数据,如果把数据的格式写到代码里面会使数据的结构不够清晰,如果数据结构变动那么改动就会比较麻烦,所以利用twi ...
Elasticsearch：如何处理 ingest pipeline 中的异常
在我之前的文章 "如何在 Elasticsearch 中使用 pipeline API 来对事件进行处理" 中,我详细地介绍了如何创建并使用一个 ingest pipeline.简 ...
【Elasticsearch 权威指南学习笔记】结构化搜索
结构化搜索(Structured search) 是指有关探询那些具有内在结构数据的过程.比如日期.时间和数字都是结构化的:它们有精确的格式,我们可以对这些格式进行逻辑操作.比较常见的操作包括比较数字 ...
Observability：Elastic 可观测性是什么？
简单地说, 可观测性将你的数据转化为可行的见解.为了有效地监控分布式系统并获得洞察力,你需要将所有可观察性数据放在一个堆栈中. 通过将应用程序.基础架构和用户数据整合到一个统一的解决方案中,实现端到端 ...
心电图数据结构化标准_自己实现一个类 JSON 数据结构
JSON,仅支持极少的数据类型,或者是为了跨平台和语言的兼容性,又或者是因为其他什么原因.总之,它虽然成为现在网络传输接口中最普遍的数据结构,但由于自身限制,在使用过程中它也给开发者带来了一些负担.诸 ...

Elasticsearch：Elastic可观测性 - 运用 pipeline 使数据结构化

数据集