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

 Elasticsearch 聚合是一项强大且灵活的功能，可让你汇总和检索有关数据的复杂见解，但是，如果你对大量数据运行复杂的聚合，则可能会耗尽内存，例如

• 你需要一个完整的特征索引而不是前 N 个项目集
• 你需要通过管道聚合对聚合结果进行排序
• 你想创建汇总表来优化查询

数据 transform 可以提供帮助！

在实际的使用案例中，比如我们有如下形式的文档：

如上所示，我们的文档不含有针对每个 entity，比如 Anna 及 Simon 的统计数据。如果我们需要得到这些 entity 的统计数据，我们可以借助于 Elasticsearch 所提供的 Transform 来完成整个操作。如果上面右下角的表格所示，我们可以针对每个 entity 得到该他们的统计数据，比如所有 Anna 文档的 Average Price 等。 在上面，我们只有3个文档，但是在实际的使用中，这个数据可能是成千上万的。我们需要借助 Elasticsearch 的 Transform 来帮我们自动生成另外一个索引。在这个索引中，我们可以生产针对每个个体形成的统计数据。依据这个数据，我们甚至可以借助机器学习的功能来判断数据的异常，比如，我们很容易发现哪个用户花去了比平常客户多几倍的价钱来购买东西，哪些客户比平常客户购买更多或更少的商品。

准备数据

在今天的练习中，我们将以 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 索引：

在上面，我们可以看到有3321个文档，而且每个文档里含有的信息如上所示。它显示了当前用户的花费信息。我们可以针对这个索引来进行搜索。这些数据在很多时候非常有用，比如在进行机器学习时，我们可以生产这样的索引，对数据进行分析。

使用 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】Tutorial: Transforming the eCommerce sample data | Elasticsearch Guide [7.7] | Elastic