---
title: Redis Hash Sink 集成
---

# Redis Hash Sink 集成

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

## 前提条件

- 已了解管道结构与创建流程，参考 [数据集成](data-integration.md)；Sink 连接器总览见 [Sink](data-integration-output.md)
- 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 · 示例场景](data-integration-output.md#示例场景)。

可参考以下参数示例：

| 参数 | 建议值 |
|---|---|
| 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](data-integration-processors-expressions.md)。

| 参数 | 支持表达式 | 说明 |
|------|------------|------|
| `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`