---
title: Processors
---

# Processors

本文说明如何在 `Processors` 与相关配置项中使用表达式：对消息做映射、过滤、补字段，以及在 Source/Sink 参数里按消息动态取值。

## 适用位置

- `Processors YAML` 中的 `mapping`（对整条消息做转换）
- Source/Sink 的动态参数（如 `topic`、`table`、`device_id`）
- 字段映射与路由键生成

## 核心概念

| 关键字 | 说明 |
|--------|------|
| `root` | 正在构造的输出消息。对 `root` 或其字段赋值，即生成新消息。 |
| `this` | 当前输入消息（只读）。用 `this.field` 访问字段。 |
| `content()` | 返回消息原始字节内容。 |

常用写法：

```text
root = this                    # 先保留原消息，再改个别字段
root.name = this.user.name     # 取嵌套字段
root.count = this.items.length()
```

## 插值语法（配置项动态取值）

部分配置字段支持 `${! ... }`：运行时按**每条消息**计算表达式，再填入该字段。

示例：

- `${! json("data.value") }`
- `${! metadata("kafka_topic") }`
- `meow-${! json("topic") }`（与字面量拼接）

需要输出字面量 `${!foo}`（不求值）时，写成：

- <code v-pre>${{!foo}}</code>

## Processors 中的 mapping

控制台 `Processors YAML` 顶层键为 `processors`，常用 `mapping` 对消息做转换：

```yaml
processors:
  - mapping: |
      root = this
      root.id = uuid_v4()
      root.processed_at = timestamp_unix()
```

### 保留原结构并追加字段

```yaml
processors:
  - mapping: |
      root = this
      root.id = uuid_v4()
      root.processed_at = timestamp_unix()
```

### 重组消息结构

```yaml
processors:
  - mapping: |
      root.user = {
        "name": this.firstName + " " + this.lastName,
        "email": this.email.lowercase()
      }
      root.topic = metadata("kafka_topic")
      root.table = metadata("kafka_topic").re_replace_all("[^a-zA-Z0-9_]", "_")
```

### 缺省值与容错

```yaml
processors:
  - mapping: |
      root = this
      root.name = this.name.or("Anonymous")
      root.score = this.score.catch(0)
```

### 删除字段或丢弃消息

```yaml
processors:
  - mapping: |
      root = this
      root.password = deleted()
      root.ssn = deleted()
```

按条件丢弃整条消息：

```yaml
processors:
  - mapping: |
      root = if this.test == true { deleted() } else { this }
```

### 条件赋值

```yaml
processors:
  - mapping: |
      root = this
      root.priority = if this.urgent == true {
        "high"
      } else if this.important == true {
        "medium"
      } else {
        "low"
      }
```

## 常用取值函数

| 函数 | 说明 | 示例 |
|------|------|------|
| `json("path")` | 从消息体 JSON 取字段，支持点路径 | `${! json("data.timestamp") }` |
| `metadata("key")` | 从消息元数据取字段 | `${! metadata("kafka_topic") }` |
| `content()` | 消息原始内容 | 在 mapping 中直接使用 |
| `uuid_v4()` | 生成 UUID | `root.id = uuid_v4()` |
| `now()` | 当前时间（RFC 3339） | `root.ts = now()` |
| `timestamp_unix()` | 当前 Unix 秒级时间戳 | `root.ts = timestamp_unix()` |
| `deleted()` | 标记删除字段或整条消息 | `root.secret = deleted()` |

## 常用字符串处理

| 方法 | 说明 | 示例 |
|------|------|------|
| `lowercase()` / `uppercase()` | 大小写转换 | `this.email.lowercase()` |
| `trim()` | 去掉首尾空白 | `this.name.trim()` |
| `replace_all(old, new)` | 普通字符串替换 | `this.path.replace_all("/", "_")` |
| `re_replace_all(pattern, replacement)` | 正则批量替换 | `metadata("kafka_topic").re_replace_all("[^a-zA-Z0-9_]", "_")` |
| `has_prefix(prefix)` / `has_suffix(suffix)` | 前/后缀判断 | `this.topic.has_prefix("iot.")` |
| `length()` | 长度 | `this.name.length()` |

常见用法：

- 主题名转表名：将 `/`、`-` 等替换为 `_`
- 路径拼接：将 `topic` 转为分层路径标识

## 类型与空值处理

| 方法 | 说明 |
|------|------|
| `or(default)` | 值为 `null` 时使用默认值 |
| `catch(fallback)` | 表达式失败时返回回退值 |
| `string()` / `number()` / `bool()` | 类型转换 |
| `exists(path)` | 判断路径是否存在 |

## 排查建议

| 现象 | 建议检查 |
|------|----------|
| 字段为空 | `json("path")` / `this.xxx` 路径是否存在 |
| 结果格式不对 | 正则或 `replace_all` 是否覆盖目标字符 |
| mapping 后消息丢失 | 是否误用了 `deleted()` 或条件丢弃 |
| 测试失败 | 先用固定值验证连接，再改回表达式 / mapping |
| 需要字面量 `${!...}` | 是否已按 <code v-pre>${{!...}}</code> 转义 |