Processors

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

适用位置

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

核心概念

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

常用写法：

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

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

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

示例：

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

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

• ${{!foo}}

Processors 中的 mapping

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

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

保留原结构并追加字段

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

重组消息结构

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_]", "_")

缺省值与容错

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

删除字段或丢弃消息

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

按条件丢弃整条消息：

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

条件赋值

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
需要字面量 ${!...}	是否已按 ${{!...}} 转义