Redis Hash Sink 集成

本文介绍如何将 FlowMQ 数据写入 Redis Hash 结构。

前提条件

• 已了解管道结构与创建流程，参考 数据集成；Sink 连接器总览见 Sink
• Redis 服务可访问
• 已规划 key 与 field 映射规则

配置步骤

1. 在“数据管道”中创建管道，完成基础信息配置。
2. 在 Source 步骤选择 Kafka（Topic：flowmq.mqtt.kafka），Processors 按需配置。
3. 在 Sink 步骤选择 Redis Hash，配置 urls、key、fields 等参数。
4. 点击“测试”验证连接。
5. 继续完成确认步骤。

配置示例

以下示例将 Kafka Topic flowmq.mqtt.kafka 中的样例消息写入目标系统。样例 JSON 与 Source 约定见 Sink · 示例场景。

可参考以下参数示例：

参数	建议值
urls	host 127.0.0.1，port 6379（可选 database）
kind	simple
master	mymaster
key	${! json("source") }（样例消息对应 1SDA526VD_POP_P1）
walk_metadata	false
walk_json_object	false

fields 字段映射示例（对应样例消息）：

field	value
value	${! json("data.value") }
type	${! json("type") }
time	${! json("time") }
timestamp	${! json("data.timestamp") }

必要表单参数

参数名后带 * 表示控制台必填。「支持表达式」为「是」时，可使用 ${! ... } 按消息动态取值，详见 Processors。

参数	支持表达式	说明
urls*	否	Redis 地址列表（每项含 host/port，可选 database）
kind*	否	Redis 模式（simple/cluster/failover）
master	否	Sentinel 主节点名。即便 kind 为 simple，也需填写非空值（如 mymaster），否则无法通过连通性校验
key*	是	Hash key
walk_metadata*	否	是否写入消息元数据字段
walk_json_object*	否	是否遍历 JSON 对象写入字段
fields*	是	显式字段映射（映射值可使用表达式）
max_in_flight*	否	并发写入上限

参数建议

• key 设计避免热点；可按消息动态生成
• fields 映射值可使用表达式，建议与业务字段一一对应

注意事项

• 注意 value 类型一致
• 即便 kind 为 simple，master 也需填写非空值；urls 中的库号字段名为 database，勿使用 db