AI 助理
备案控制台
开发者社区 云原生 云原生可观测 文章 正文

【最佳实践】ingest对异源数据结构化处理，并由Elastic Stack实现可观测性分析

2020-07-01 1637
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
性能测试 PTS，5000VUM额度
可观测监控 Prometheus 版，每月50GB免费额度
可观测链路 OpenTelemetry 版，每月50GB免费额度
简介： 本文将讲述如何运用Elasticsearch的 ingest 节点实现数据结构化，并对数据进行处理。

数据集

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

image.png

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

结构化数据

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

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

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

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

image.png

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

image.png

我们可以使用 Logstash 和 Ingest node 来对我们的数据进行处理。

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

• split
• dissect
• kv
• grok
• ...

Ingest pipelines

一个 Elasticsearch pipeline 是一组 processors:

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

image.png

定义一个 Elasticsearch 的 ingest pipeline

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

image.png

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

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

image.png

上面的运行的结果是：


{
  "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"
        }
      }
    }
  ]
}

如何使用 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 来结构化这个消息：

image.png

正如我们上面显示的那样，我们想提取上面用红色标识的部分，但是我们并不需要信息中中括号【 及 】。我可以使用 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"
        }
      }
    }
  ]
}

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 里的结构化数据。我们可以在 Kibana 中打入如下的命令来查询现有的预设的 grok pattern:

GET /_ingest/processor/grok

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

image.png

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"
        }
      }
    }
  ]
}
声明：本文由原文作者“ Elastic 中国社区布道师——刘晓国”授权转载，对未经许可擅自使用者，保留追究其法律责任的权利。

image.png

【阿里云Elastic Stack】100%兼容开源ES，独有9大能力，提供免费 X-pack服务（单节点价值$6000）

相关活动

更多折扣活动，请访问阿里云 Elasticsearch 官网

阿里云 Elasticsearch 商业通用版，1核2G ,SSD 20G首月免费
阿里云 Logstash 2核4G首月免费

image.png

image.png

文章标签：
检索分析服务 Elasticsearch版
应用服务中间件
数据采集
索引
固态存储
nginx
Apache
API
数据库
关键词：
数据结构stack
来 源：
开发者社区 > 大数据与机器学习 > 检索分析服务 Elasticsearch版 > 文章 > 正文
相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群，使用DataWorks将MySQL数据同步至Elasticsearch，体验多条件检索效果，简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示，Elasticsearch是最受欢迎的企业搜索引擎，其次是Apache Solr（也是基于Lucene）。 ElasticSearch的实现原理主要分为以下几个步骤： 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词，将其权重和分词结果一并存入数据 当用户搜索数据时候，再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索，具有接近实时的搜索，并支持多租户。
工程师甲
目录
相关文章
GuGuStudy
|
1月前
|
C语言
数据结构------栈（Stack）和队列（Queue）
数据结构------栈（Stack）和队列（Queue）
GuGuStudy
19 0
源码星辰
|
3月前
|
存储
数据结构——栈（Stack）
栈（Stack）是一种常见且重要的数据结构，它遵循后进先出（Last-In-First-Out, LIFO）的原则，即最后加入的元素会是第一个被移除的。
源码星辰
43 4
jjczqyjxgdegi
|
5月前
|
存储 人工智能 程序员
技术心得记录：堆(heap)与栈(stack)的区别
技术心得记录：堆(heap)与栈(stack)的区别
jjczqyjxgdegi
40 0
ali_Acat-45511
|
5月前
|
C语言 C++
【数据结构】C语言实现：栈（Stack）与队列（Queue）
【数据结构】C语言实现：栈（Stack）与队列（Queue）
ali_Acat-45511
62 0
炫酷的伊莉娜
|
6月前
|
存储
【数据结构】栈（Stack）的实现 -- 详解
【数据结构】栈（Stack）的实现 -- 详解
炫酷的伊莉娜
68 0
fly丶知秋
|
存储 canal 算法
数据结构之Stack | 让我们一块来学习数据结构
数据结构之Stack | 让我们一块来学习数据结构
fly丶知秋
58 0
小白在努力jy
|
测试技术 C语言
[数据结构 -- C语言] 栈(Stack)
[数据结构 -- C语言] 栈(Stack)
小白在努力jy
110 0
张思睿ZGQ
|
6月前
|
算法 安全 Java
【数据结构与算法】6、栈（Stack）的实现、LeetCode：有效的括号
【数据结构与算法】6、栈（Stack）的实现、LeetCode：有效的括号
张思睿ZGQ
43 0
tao滔不绝
|
6月前
数据结构 模拟实现Stack栈（数组模拟）
数据结构 模拟实现Stack栈（数组模拟）
tao滔不绝
66 0
算精通
递归工作栈（Recursive Workstation Stack）
递归工作栈（Recursive Workstation Stack）是一种在计算机程序中实现递归计算的机制，通过使用栈来跟踪递归调用的过程，从而实现对复杂问题的求解。递归工作栈在解决具有自相似结构的问题时非常有用，例如计算斐波那契数列、解决迷宫问题等。
算精通
271 9
云原生

云原生可观测

热门文章

最新文章

  • 1
    扩展AlertManager集成钉钉助力Istio on ACK可观测性监控能力
  • 2
    MES系统软件体系架构及应用
  • 3
    这才是可观测告警运维平台——20个SLS告警运维场景
  • 4
    问题盘点｜使用 Prometheus 监控 Kafka，我们该关注哪些指标
  • 5
    千万级可观测数据采集器--iLogtail代码完整开源
  • 6
    Linux调试
  • 7
    可观测告警运维系统调研——SLS告警与多款方案对比
  • 8
    可观测平台下告警降噪实践——GOPS分享
  • 9
    Grafana Plugin: 支持 jsx 实时渲染的 grafana 面板插件
  • 10
    一文了解什么是CNAPP
  • 1
    Linux调试
    20
  • 2
    （H5-Web3D-ThreeJS）在网页三维CAD中绘制窗户模型
    39
  • 3
    【云故事探索】NO.8：揭秘餐饮行业龙头 SaaS 厂商神州商龙的全栈可观测实践
    99
  • 4
    【云故事探索】NO.8：揭秘餐饮行业龙头 SaaS 厂商神州商龙的全栈可观测实践
    71
  • 5
    一文了解什么是CNAPP
    350
  • 6
    测试问题之可观测性的本质是什么，SLS在可观测性领域采取了什么样的策略
    48
  • 7
    grafana展示的CPU利用率与实际不符的问题探究
    239
  • 8
    SRE 排障利器，接口请求超时试试 httpstat
    95
  • 9
    Grafana Plugin: 支持 jsx 实时渲染的 grafana 面板插件
    284
  • 10
    如何在DocuSign中设置PKCE（Proof Key for Code Exchange）
    178

    • 相关课程

    更多
  • Go语言核心编程 - 数据结构和算法
  • 如何使用Kubernetes监控定位慢调用
  • 日志服务 SLS 可观测数据分析平台介绍

    • 相关电子书

    更多
  • 如何使用Tair增强数据结构构建丰富在线实时场景
  • Apache Flink 流式应用中状态的数据结构定义升级
  • Elastic Stack 运维数据分析从0到1

    • 相关实验场景

    更多
  • 使用检索分析服务Elasticsearch版实现基础检索
  • Elastic Stack 8 快速上手
  • 使用阿里云Elasticsearch快速搭建可观测系统
    • 下一篇
    DataWorks售前咨询