在 Elasticsearch 中，Transform 是实现 实时聚合、数据透视、ETL（Extract-Transform-Load） 的强大功能。它能将原始数据流按维度聚合，并将结果写入目标索引，适用于：

• 实时大屏监控
• 用户行为分析
• 业务指标报表
• 降频存储（高频原始数据 → 低频汇总）

本文为你设计一套 Elasticsearch Transform 实时聚合方案，支持低延迟、高吞吐、可扩展的实时数据分析。

---

一、目标与核心需求

目标	说明
⚡ 低延迟	数据变更到聚合结果可见 < 30 秒
📈 高吞吐	支持每秒数万文档聚合
🧱 可扩展	支持多维度、多聚合类型
🔄 增量更新	基于 checkpoint 增量处理
💾 持久化结果	聚合结果可被 Kibana 查询、告警
📊 可观测性	监控延迟、失败率、资源使用

---

二、系统架构设计

+------------------+     +---------------------+
| 业务系统 / 日志    | --> | Kafka / Filebeat    |
+------------------+     +----------+----------+
                                    |
                                    v
                     +--------------------------+
                     | Elasticsearch Raw Index  |
                     | - logs-raw, user-events  |
                     +------------+-------------+
                                  |
                                  v
                     +--------------------------+
                     | Transform Job            |
                     | - pivot / latest         |
                     | - sync: time-based       |
                     +------------+-------------+
                                  |
                                  v
                     +--------------------------+
                     | Aggregated Index         |
                     | - user-summary           |
                     | - hourly-metrics         |
                     +------------+-------------+
                                  |
                                  v
                     +--------------------------+
                     | Kibana / Dashboard       |
                     | - 可视化、告警、探索     |
                     +--------------------------+

✅ 原始数据写入源索引，Transform 实时聚合到目标索引。

---

三、Transform 核心概念

1. 两种模式

模式	说明	适用场景
Pivoted	按字段分组 + 聚合（行列转换）	统计报表、Dashboard
Latest	按字段取最新文档	同步最新状态（如用户资料）

---

2. 增量同步机制

• Checkpoint：记录已处理的文档位置；
• Sync：支持 time 或 seq_no 增量同步；
• Frequency：轮询间隔（如 30s）。

---

四、实时聚合方案设计

方案 1：用户行为汇总（Pivoted）

场景

• 统计每个用户的访问次数、最后登录时间、消费总额。

Step 1：源数据

{
  "user_id": "u123",
  "action": "purchase",
  "amount": 29900,
  "timestamp": "2024-06-01T10:00:00Z"
}

Step 2：创建 Transform

PUT _transform/user-summary-transform
{
  "source": {
    "index": "user-events-*"
  },
  "pivot": {
    "group_by": {
      "user_id": {
        "terms": {
          "field": "user_id"
        }
      }
    },
    "aggregations": {
      "total_amount": {
        "sum": {
          "field": "amount"
        }
      },
      "visit_count": {
        "value_count": {
          "field": "action"
        }
      },
      "last_seen": {
        "max": {
          "field": "timestamp"
        }
      },
      "top_action": {
        "terms": {
          "field": "action",
          "size": 1
        }
      }
    }
  },
  "description": "用户行为汇总",
  "dest": {
    "index": "user-summary"
  },
  "frequency": "30s",
  "sync": {
    "time": {
      "field": "timestamp",
      "delay": "60s"
    }
  }
}

• frequency: 30s：每 30 秒检查增量；
• delay: 60s：容忍数据延迟。

Step 3：启动 Transform

POST _transform/user-summary-transform/_start

结果文档：

{
  "user_id": "u123",
  "total_amount": 59800,
  "visit_count": 5,
  "last_seen": "2024-06-01T18:00:00Z",
  "top_action": { "buckets": [ { "key": "view", "doc_count": 3 } ] }
}

---

方案 2：每小时指标聚合（Time-based）

场景

• 每小时统计订单量、销售额、地域分布。

Transform 配置

PUT _transform/hourly-metrics
{
  "source": { "index": "orders-*" },
  "pivot": {
    "group_by": {
      "hour": {
        "date_histogram": {
          "field": "order_time",
          "fixed_interval": "1h"
        }
      },
      "region": {
        "terms": { "field": "region" }
      }
    },
    "aggregations": {
      "order_count": { "value_count": { "field": "order_id" } },
      "revenue": { "sum": { "field": "amount" } },
      "avg_amount": { "avg": { "field": "amount" } }
    }
  },
  "dest": { "index": "metrics-hourly" },
  "frequency": "60s",
  "sync": {
    "time": {
      "field": "order_time",
      "delay": "5m"
    }
  }
}

聚合结果可用于 Kibana 实时大屏。

---

方案 3：最新状态同步（Latest）

场景

• 将用户最新资料同步到搜索索引。

PUT _transform/user-latest-transform
{
  "source": { "index": "user-changelog" },
  "latest": {
    "unique_key": [ "user_id" ],
    "sort": "timestamp"
  },
  "dest": { "index": "user-search" },
  "frequency": "15s"
}

每 15 秒检查是否有新用户变更，更新 user-search。

---

五、性能优化建议 ✅

场景	建议
大基数 group_by	使用 composite 聚合避免内存溢出
高频更新	降低 frequency（如 15s）
数据延迟	增加 delay 容忍乱序
聚合字段	使用 keyword 而非 text
目标索引	提前定义 Mapping，避免动态映射
资源隔离	将 Transform 节点与数据节点分离

---

六、监控与可观测性

1. 关键监控指标

指标	说明
transform.checkpoint.duration.ms	单次执行耗时
transform.indexing.time.ms	写入目标索引耗时
transform.search.time.ms	查询源数据耗时
transform.processing.time.ms	聚合处理耗时
transform.documents.processed	处理文档数
transform.search.failures	搜索失败次数

通过 GET _transform/_stats 获取。

---

2. 告警规则

• checkpoint.duration > 10s（处理过慢）
• search.failures > 0
• documents.processed == 0 for 5m（停滞）

---

七、容错与恢复

机制	说明
Checkpoint 持久化	失败后从上次位置恢复
幂等写入	目标索引使用 doc_as_upsert
自动重启	Transform 失败后自动重试
手动干预	可暂停、重置、重新启动

---

八、扩展建议

场景	建议方案
多源聚合	使用 source.index 支持通配符（logs-*）
复杂逻辑	在 pipeline 中使用 script 预处理
冷热分离	目标索引使用 ILM 自动 rollover 和 shrink
权限控制	限制 Transform 只能访问特定索引
测试环境	使用 preview API 验证 Transform 输出

---

九、总结：Transform 实时聚合 checklist ✅

项目	是否完成
选择合适模式（pivoted/latest）	✅
设置合理的 frequency 和 delay	✅
源索引使用时间字段同步	✅
目标索引提前建模	✅
监控 checkpoint 延迟	✅
告警机制	✅
性能优化（避免大 group_by）	✅
权限与安全	✅