---
title: DolphinDB Sink 集成
---

# DolphinDB Sink 集成

本文介绍如何将 FlowMQ 数据写入 DolphinDB。

## 前提条件

- 已了解管道结构与创建流程，参考 [数据集成](data-integration.md)；Sink 连接器总览见 [Sink](data-integration-output.md)
- DolphinDB 服务可访问
- 已具备目标库表写入权限

## 配置步骤

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

## 配置示例

以下示例将 Kafka Topic `flowmq.mqtt.kafka` 中的样例消息写入目标系统。样例 JSON 与 Source 约定见 [Sink · 示例场景](data-integration-output.md#示例场景)。

可参考以下参数示例：

| 参数 | 建议值 |
|---|---|
| urls | `127.0.0.1:8848` |
| username | `admin` |
| password | `123456` |
| directory | `dfs://flowmq` |
| table | `mqtt_kafka_events` |
| memory_cache_size | `10000` |
| subscribe.batch_size | `1` |
| subscribe.batch_interval | `1`（单位：秒） |

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

| name | data_type | is_partition | is_sort | value | value_source |
|------|-----------|--------------|---------|-------|--------------|
| mpCode | STRING | false | true | `source` | `json path` |
| valueTime | TIMESTAMP | true | false | `data.timestamp` | `json path` |
| value | FLOAT | false | false | `data.value` | `json path` |
| unitCode | STRING | false | false | `type` | `json path` |

> 每列指定 `value_source`（默认 `direct`）。上表示例为 `json path`。分区列 `valueTime` 映射 `data.timestamp`（毫秒），请确保落在库的日期分区范围内。样例 `data.value` 为字符串时也可写入 `FLOAT` 列。

## value_source 说明（DolphinDB）

`columns` 中仅 `value` 受 `value_source` 控制；`name`、`data_type`、`is_partition`、`is_sort` 不受影响。

| value_source | `value` 填写内容 | 运行时行为 |
|--------------|------------------|------------|
| `direct` | 固定字面量，或 `${! ... }` 表达式 | 不含 `${!}` 作为静态值；含 `${! ... }` 时按每条消息执行插值 |
| `meta` | 元数据键名（字符串） | 从消息 metadata 取对应键，键不存在时报错 |
| `json path` | JSON 点路径（如 `source`、`data.value`、`data.sensors[0].temp`） | 将消息解析为结构化 JSON 后按路径取值，路径不存在或值为 `null` 时报错（`[N]` 解析为 `.N`） |

使用 Kafka Source（`kafka_franz`）时，可用的 metadata 键包括：`kafka_key`、`kafka_topic`、`kafka_partition`、`kafka_offset`、`kafka_timestamp_unix`、`kafka_tombstone_message`。

## 必要表单参数

参数名后带 `*` 表示控制台必填。「支持表达式」为「是」时，可使用 `${! ... }` 按消息动态取值，详见 [Processors](data-integration-processors-expressions.md)。

| 参数 | 支持表达式 | 说明 |
|------|------------|------|
| `urls*` | 否 | DolphinDB 地址列表（如 `127.0.0.1:8848`） |
| `username*` | 否 | 用户名 |
| `password*` | 否 | 密码 |
| `directory*` | 否 | 目标库路径（如 `dfs://flowmq`） |
| `table*` | 否 | 目标表名 |
| `memory_cache_size*` | 否 | 流表缓存大小 |
| `subscribe*` | 否 | `batch_size` / `batch_interval`（秒） |
| `columns*` | 是 | 列映射（含 `value_source`） |
| `max_in_flight*` | 否 | 并发写入上限 |

## 参数建议

- `subscribe.batch_interval` 单位为秒；需要尽快看到写入结果时，可将 `batch_size` 设为 `1`
- `columns` 中每列指定 `value_source`（默认 `direct`）；从样例 JSON 取字段可优先用 `json path`
- 分区字段取值须落在库自动生成的日期分区范围内

## 注意事项

- 完成连通性测试后，建议再写入样例消息确认数据已落入目标表
- 请确认 DolphinDB 版本与 FlowMQ 内置客户端兼容