Elasticsearch的Transforms介绍

Transforms 使您能够从 Elasticsearch 索引中检索信息，对其进行转换并将其存储在另一个索引中。 使您能够透视数据并创建以实体为中心的索引，这些索引可以汇总实体的行为。 这会将数据组织成易于分析的格式。让我们使用Kibana示例数据来演示如何使用变换来透视和汇总数据。

准备数据

在今天的练习中，我们将以 eCommerce 订单的样本例子来做练习。首先，我们来把数据导入到 Elasticsearch 中：

 

 点击 Add data 按钮：

 这样我们就完成了 eCommerce 数据的导入。如果您还不熟悉kibana_sample_data_ecommerce索引，请使用Kibana中的 Revenue仪表板浏览数据。 考虑一下你可能想从此电子商务数据中获得什么见解:

 

 使用各种选项进行分组和汇总

透视数据涉及使用至少一个字段对其进行分组并应用至少一项聚合。 你可以预览转换后的数据，然后继续进行操作！

例如，你可能想按产品ID对数据进行分组，并计算每种产品的销售总数及其平均价格。 另外，你可能希望查看单个客户的行为，并计算每个客户总共花费了多少以及他们购买了多少种不同类别的产品。 或者，你可能需要考虑货币或地理位置。 转换和解释这些数据最有趣的方式是什么？

下面，我们来做一个练习。打开 Kibana：

 我们点击上面的 Create your first transform：

 点击上面的 [eCommerce] Orders:

 我们在上面进行一些感兴趣的项进行选择：

 我们在屏幕的右方可以看到 Transform pivot preview。它显示的就像是一个表格的形式，而里面的数据是我们原始的数据里没有的信息，比如它含有 products.quantity 的总和等。

我们点击当前页面的 Next 按钮：

 点击上面的 Next：

 我们点击上面的 Create and start 按钮：

 上面的 progress 显示 transform 的进度。已经完成100%了。点击上面的红色框，我们就可以回到 Transform 的管理页面了。

 在上面，它显示我们的 Transform 已经完成了。状态是 stopped。这是因为我们的数据量还不是很大的缘故。我们点击上面的向下的扩展箭头：

我们可以看到所有的转换的细节。

我们接下来到 Discover 中去查看我们最新生产的一个索引：ecommerce-customer-sales

 我们选中 ecommerce-customer-sales 索引：

 使用 API 来完成 transform

上面我们使用了 Kibana 中的 GUI 来完成这个工作。实际上我们也可以使用 API 的方式来完成这个工作。

我们先定义一个需要被使用的 pipeline:

PUT _ingest/pipeline/add_timestamp_pipeline
{
  "description": "Adds timestamp to documents",
  "processors": [
    {
      "script": {
        "source": "ctx['@timestamp'] = new Date().getTime();"
      }
    }
  ]
}

执行上面的指令。然后执行下面的指令：

PUT _transform/ecommerce_transform
{
  "source": {
    "index": "kibana_sample_data_ecommerce",
    "query": {
      "term": {
        "geoip.continent_name": {
          "value": "Asia"
        }
      }
    }
  },
  "pivot": {
    "group_by": {
      "customer_id": {
        "terms": {
          "field": "customer_id"
        }
      }
    },
    "aggregations": {
      "max_price": {
        "max": {
          "field": "taxful_total_price"
        }
      }
    }
  },
  "description": "Maximum priced ecommerce data by customer_id in Asia",
  "dest": {
    "index": "kibana_sample_data_ecommerce_transform",
    "pipeline": "add_timestamp_pipeline"
  },
  "frequency": "5m",
  "sync": {
    "time": {
      "field": "order_date",
      "delay": "60s"
    }
  }
}

在上面，我们定义了一个 持续 transform，也就是说每隔5分钟的时间，它会检查最新的数据，并进行转换。这个在 frequency 里有定义。当我们执行上面的命令后，我们可以在 transform 的管理页面看到：

 我们看到一个新的 Transform 已经生产，而且它是一个 continuous 的 transform。我们可以点击 Start 按钮来执行它。我们也可以使用如下的 API 来启动这个 transform：

POST _transform/ecommerce_transform/_start

当我们执行完上面的命令后，我们再次查看 transform 的管理页面：

 从上面我们可以看出来这个 transform 已经被启动，而且是一种在运行的状态。我们可以点击 Stop 来停止这个 transform，如果我们不想运行的话。

由于上面的命令没有为新创建的索引 kibana_sample_data_ecommerce_transform 创建一个index pattern，我们需要自己手动来创建一个 index pattern。等我们创建完后，打开 Discover 来查看新的 transform 索引：

 

 从上面我们可以看到有 13 个文档，这是因为我们只关心 Asia 的数据。所有的数据是以 customer_id 来分组的。它显示了这个 customer 的最大价格。在上面我们看到我们也有一个通过 pipeline 写入的当前时间。

有兴趣的开发者，可以尝试写入一个新的文档到 kibana_sample_data_ecommerce 索引，并且是 Asia 的，我们可以看看是否有多一个文档在 kibana_sample_data_ecommerce_transform 索引中。

我们可以通过如下的 API 来删除这个 transform:

POST _transform/ecommerce_transform/_stop
DELETE _transform/ecommerce_transform

参考：

【1】https://www.elastic.co/guide/en/elasticsearch/reference/7.7/ecommerce-transforms.html

原文链接：https://blog.csdn.net/ubuntutouch/java/article/details/106261158