Skip to Content
Docs问题排查809 connector 数据流链路排查手册

版本: v1.0 修改日期: 2026-05-29 版本： v1.1 适用场景： 采集数据中断、数据缺失、数据未入库、MQTT 无推送、Kafka 无数据

1 整体排查流程

排查按数据流链路逐层推进：先确认下位协议连通，再查 source 采集 → common 处理 → business 标准化写出 → 最终落库与推送。每层输出是下一层的输入，上游正常则向下追。以下统称**「数据对接容器」**（connector），按阶段分为 source / common / business 三个阶段。

图 1 数据流链路排查流程

步骤操作目标对应章节
1确认下位协议连通性第 2 章
2确认 source 容器采集正常第 3 章
3确认 common 容器处理正常第 4 章
4确认 business 容器标准化写出正常第 5 章
5需深入排查 Kafka 各阶段 Topic 时使用第 6 章
6验证数据库落库情况第 7 章
7验证 MQTT 推送第 8 章

2 输入协议层排查

确认下位设备与 viSCADA 平台之间的协议连通性。

  1. 确认设备类**「通讯协议」**已正确选择（Modbus / S7 / OPC-UA / HTTP / MQTT / Kafka / 数据库）。
  2. 确认设备**「通讯地址」**填写无误（IP / 端口 / 从站地址等）。
  3. 确认设备在线状态（网络可达、设备未断电）。
  4. 若为数据库协议，确认 DSN 连接字符串格式正确、数据库端白名单已放通。

ℹ️ 七种协议的详细配置参数见《数据对接配置手册》。

3 source 容器排查

目的： 确认 source 容器运行正常，采集脚本有日志输出，Kafka device.${TENANT_ID}.data Topic 有数据流入。

3.1 配置日志脚本

在**「设备类」**中添加日志脚本，用于在 Telegraf pipeline 中拦截采集数据并打印到日志，从而确认 source 容器是否收到了设备数据。

  1. 进入**「设备类」** → 「脚本」。

  2. 点击**「添加脚本」**，输入以下脚本内容：

    [[processors.starlark]]
    namepass = ["${TEMPLATE_NAME}"]
    alias = "${TEMPLATE_NAME}@starting"
    order = ${order@54}
    source = '''
load("logging.star", "log")
load("json.star", "json")
def apply(metric):
    log.info("metricStart:"+json.encode(metric))
    return metric
'''

  3. 点击**「√」**保存。如图 2。

    图 2 在设备类中添加日志脚本

ℹ️ 脚本作用：收到一条采集数据时，打印 metricStart:{...} 到日志。若日志中无此输出，说明 source 容器未收到数据。${TEMPLATE_NAME} 替换为 Telegraf 配置中对应 name 字段的实际值；order@54 表示插入到 pipeline 第 54 位，按项目约定调整。

3.2 在服务器操作系统中查询 source 容器日志

source 容器命名格式：connector-${TENANT_ID}-source-0-xxxxxxxxxx-xxxxx。

租户 IDPod 名称示例
90connector-90-source-0-55d68b5978-882jd
95connector-95-source-0-7ddc6ff6cb-xz9b7
123456connector-123456-source-0-7689986d67-jhctm

  1. 定位目标容器，确认状态（Running）、重启次数及所在节点：

    kubectl -n prod get pods -o wide | grep source
#       |        |         |         └── 过滤名称含 source 的容器
#       |        |         └── 宽格式输出（含 IP、节点等信息）
#       |        └── 列出所有容器
#       └── 指定命名空间为 prod

    执行后输出示例：

    NAME                                          READY   STATUS    RESTARTS   AGE   IP            NODE
connector-90-source-0-55d68b5978-882jd        1/1     Running   0          2d    10.0.1.23     node-01
    输出字段说明
    NAME容器名称，用于后续日志查询命令
    READY就绪状态，1/1 表示正常
    STATUS运行状态，应为 Running；若为 CrashLoopBackOff 或 Error 则需进一步排查
    RESTARTS重启次数，次数持续增加说明容器异常
    NODE所在服务器节点名称

  2. 查看容器日志：

    # 查看全量日志
kubectl -n prod logs connector-90-source-0-55d68b5978-882jd
#       |        |      └── 容器名称（第 1 步获取）
#       |        └── 查看日志
#       └── 指定命名空间为 prod
 
# 实时跟踪（Ctrl+C 退出）
kubectl -n prod logs -f connector-90-source-0-55d68b5978-882jd
#                         └── 持续输出新日志

    💡 日志量过大时追加 --tail=200 --since=1h 缩小范围：

    kubectl -n prod logs --tail=200 --since=1h <pod-name>
#                    |          └── 只看最近 1 小时内的日志
#                    └── 只取最后 200 行
kubectl -n prod logs -f --tail=200 --since=1h <pod-name>

  3. 按需深入过滤，如图 3：

    # 仅查看报错日志
kubectl -n prod logs -f connector-90-source-0-55d68b5978-882jd | grep 'E!'
#                                                                  └── 过滤含 E! 的报错行
 
# 按设备类标识过滤
kubectl -n prod logs -f connector-90-source-0-55d68b5978-882jd | grep '9DQpiyceGi'
#                                                                  └── 替换为实际设备类标识
 
# 排除噪音 + 按设备过滤
kubectl -n prod logs -f connector-90-source-0-55d68b5978-882jd | grep -v '10000' | grep 'JuIzTd8MQ0'
#                                                                  |               └── 再按设备标识过滤
#                                                                  └── 排除含 10000 的噪音行
 
# 容器状态异常时查看事件
kubectl -n prod describe pod connector-90-source-0-55d68b5978-882jd
#       |        |           └── 容器名称
#       |        └── 查看容器详情与事件
#       └── 指定命名空间为 prod
kubectl -n prod get events --sort-by=.lastTimestamp | grep connector-90-source-0-55d68b5978-882jd
#       |        |          └── 按时间排序                └── 过滤该容器相关事件
#       |        └── 列出事件
#       └── 指定命名空间为 prod

    图 3 命令行查看采集数据

4 common 容器排查

目的： 确认 common 容器正常消费 device.${TENANT_ID}.data，并已将数据写入 Kafka original.${TENANT_ID}.data Topic。

4.1 在服务器操作系统中查询 common 容器日志

common 容器命名格式：connector-${TENANT_ID}-common-xxxxxxxxxx-xxxxx。

租户 IDPod 名称示例
90connector-90-common-649f8df58d-8kqd4
95connector-95-common-69d4dc74b-spmp6
123456connector-123456-common-6d8ccd7df5-wgflt

  1. 定位目标容器，确认状态（Running）、重启次数及所在节点：

    kubectl -n prod get pods -o wide | grep common
#       |        |         |         └── 过滤名称含 common 的容器
#       |        |         └── 宽格式输出（含 IP、节点等信息）
#       |        └── 列出所有容器
#       └── 指定命名空间为 prod

  2. 查看容器日志：

    # 查看全量日志
kubectl -n prod logs connector-90-common-649f8df58d-8kqd4
#       |        |      └── 容器名称（第 1 步获取）
#       |        └── 查看日志
#       └── 指定命名空间为 prod
 
# 实时跟踪（Ctrl+C 退出）
kubectl -n prod logs -f connector-90-common-649f8df58d-8kqd4
#                         └── 持续输出新日志

    💡 日志量过大时追加 --tail=200 --since=1h 缩小范围：

    kubectl -n prod logs --tail=200 --since=1h <pod-name>
#                    |          └── 只看最近 1 小时内的日志
#                    └── 只取最后 200 行
kubectl -n prod logs -f --tail=200 --since=1h <pod-name>

  3. 按需深入过滤：

    # 仅查看报错日志
kubectl -n prod logs -f connector-90-common-649f8df58d-8kqd4 | grep 'E!'
#                                                               └── 过滤含 E! 的报错行
 
# 按设备类标识过滤
kubectl -n prod logs -f connector-90-common-649f8df58d-8kqd4 | grep '9DQpiyceGi'
#                                                               └── 替换为实际设备类标识
 
# 容器状态异常时查看事件
kubectl -n prod describe pod connector-90-common-649f8df58d-8kqd4
#       |        |           └── 容器名称
#       |        └── 查看容器详情与事件
#       └── 指定命名空间为 prod
kubectl -n prod get events --sort-by=.lastTimestamp | grep connector-90-common-649f8df58d-8kqd4
#       |        |          └── 按时间排序                └── 过滤该容器相关事件
#       |        └── 列出事件
#       └── 指定命名空间为 prod

4.2 验证输出

  1. 进入 Kafka broker，消费 original.${TENANT_ID}.data Topic 验证数据（替换 ${TENANT_ID} 为实际租户 ID）：

    kubectl exec -n kafka -it kafka-0 -- bash
#       |      |         |   |       └── 进入容器执行 bash
#       |      |         |   └── 分隔符（kubectl 参数结束）
#       |      |         └── 容器名称
#       |      └── 交互式终端
#       └── kafka 命名空间
cd /opt/bitnami/kafka/bin/
sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic original.1.data --max-messages 5
#   |       |                 └── 只取 5 条后退出
#   |       └── 目标 Topic（替换 TENANT_ID）
#   └── Kafka broker 地址

  2. 验证 MQTT 原始推送，详见第 8 章。

5 business 容器排查

目的： 确认 business 容器正常消费 Kafka original.${TENANT_ID}.data，各业务域 Kafka Topic 有数据流入流出。

5.1 在服务器操作系统中查询 business 容器日志

business 容器命名格式：connector-${TENANT_ID}-business-(gongye|minyong|pdos|iems)-xxxxxxxxxx-xxxxx，每个租户 4 个实例。

示例（租户 90）：

业务域Pod 名称示例
gongyeconnector-90-business-gongye-6bcff4548b-vnhp7
minyongconnector-90-business-minyong-578d8bc468-vrksk
pdosconnector-90-business-pdos-6795b6b686-99kcj
iemsconnector-90-business-iems-5657949cbf-d9kk5

  1. 定位容器，确认 4 个实例均为 Running：

    kubectl -n prod get pods -o wide | grep business
#       |        |         |         └── 过滤名称含 business 的容器
#       |        |         └── 宽格式输出（含 IP、节点等信息）
#       |        └── 列出所有容器
#       └── 指定命名空间为 prod

  2. 根据业务线选择对应容器查看日志（以租户 90 工业线为例，其余替换容器名称即可）：

    # 工业 (gongye)
kubectl -n prod logs connector-90-business-gongye-6bcff4548b-vnhp7
#       |        |      └── 容器名称（第 1 步获取）
#       |        └── 查看日志
#       └── 指定命名空间为 prod
kubectl -n prod logs -f connector-90-business-gongye-6bcff4548b-vnhp7
#                         └── 持续输出新日志（Ctrl+C 退出）
 
# 民用 (minyong)
kubectl -n prod logs connector-90-business-minyong-578d8bc468-vrksk
 
# 光伏/配电 (pdos)
kubectl -n prod logs connector-90-business-pdos-6795b6b686-99kcj
 
# 综合能源 (iems)
kubectl -n prod logs connector-90-business-iems-5657949cbf-d9kk5

    💡 日志量过大时追加 --tail=200 --since=1h 缩小范围：

    kubectl -n prod logs --tail=200 --since=1h <pod-name>
#                    |          └── 只看最近 1 小时内的日志
#                    └── 只取最后 200 行
kubectl -n prod logs -f --tail=200 --since=1h <pod-name>

  3. 按需深入过滤（以 gongye 为例）：

    # 仅查看报错日志
kubectl -n prod logs -f connector-90-business-gongye-6bcff4548b-vnhp7 | grep 'E!'
#                                                                         └── 过滤含 E! 的报错行
 
# 按设备类标识过滤
kubectl -n prod logs -f connector-90-business-gongye-6bcff4548b-vnhp7 | grep '9DQpiyceGi'
#                                                                         └── 替换为实际设备类标识
 
# 容器状态异常时查看事件
kubectl -n prod describe pod connector-90-business-gongye-6bcff4548b-vnhp7
#       |        |           └── 容器名称
#       |        └── 查看容器详情与事件
#       └── 指定命名空间为 prod
kubectl -n prod get events --sort-by=.lastTimestamp | grep connector-90-business-gongye-6bcff4548b-vnhp7
#       |        |          └── 按时间排序                └── 过滤该容器相关事件
#       |        └── 列出事件
#       └── 指定命名空间为 prod

5.2 验证 Kafka 业务 Topic

  1. 消费各业务域私有 Kafka Topic，确认有新消息写入：

    # 工业 (gongye)
sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic gongye.1.data --max-messages 3
#   |       |               └── 只取 3 条后退出
#   |       └── 目标 Topic（替换 TENANT_ID）
#   └── Kafka broker 地址
 
# 光伏/配电 (pdos)
sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic pdos.1.data --max-messages 3

  2. 验证 MQTT 标准化推送，详见第 8 章。

6 Kafka 主题排查

数据流链路中 Kafka 承担三阶段传输，每阶段的 Topic 不同。日常排查跟随第 3～5 章中各容器的验证步骤即可；当需要独立深入排查 Kafka 时，使用本章命令逐阶段消费验证。

阶段生产者Kafka Topic消费者
1source 容器device.${TENANT_ID}.datacommon 容器
2common 容器original.${TENANT_ID}.databusiness 容器
3business 容器gongye.${TENANT_ID}.data / minyong.${TENANT_ID}.data / pdos.${TENANT_ID}.data / iems.${TENANT_ID}.databusiness 容器写出阶段

6.1 进入 Kafka broker

  1. 查看 Kafka 命名空间下所有容器，确认所有 broker 状态为 Running：

    kubectl get pods -n kafka
#       |         └── 指定命名空间为 kafka
#       └── 列出所有容器

  2. 进入 Kafka broker 容器：

    kubectl exec -n kafka -it kafka-0 -- bash
#       |      |         |   |       └── 进入容器执行 bash
#       |      |         |   └── 分隔符（kubectl 参数结束）
#       |      |         └── 容器名称
#       |      └── 交互式终端
#       └── kafka 命名空间
# 若镜像不带 bash，改用：kubectl exec -n kafka -it kafka-0 -- sh

  3. 切换到 Kafka CLI 脚本目录：

    cd /opt/bitnami/kafka/bin/   # Bitnami 镜像
# cd /opt/kafka/bin/          # Apache 官方镜像

6.2 列出与查看 Topic

  1. 列出所有 Topic，确认目标 Topic 是否存在：

    sh kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list
#                    |                                └── 列出所有 Topic
#                    └── Kafka broker 地址

  2. 查看 Topic 分区、副本与 leader 状态：

    sh kafka-topics.sh --bootstrap-server 127.0.0.1:9092 \
    --describe --topic device.1.data
#   |           └── 目标 Topic 名称（替换 TENANT_ID）
#   └── 查看 Topic 详情（分区数、副本分布、leader）

6.3 消费 Topic 验证数据

  1. 拉取最近几条数据，确认 Topic 有数据：

    sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic device.1.data --max-messages 3
#   |       |               └── 只取 3 条后退出
#   |       └── 目标 Topic（替换 TENANT_ID）
#   └── Kafka broker 地址

  2. 实时消费并按关键词过滤：

    sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic device.1.data | grep '48985'
#                           └── 按关键字过滤（替换为实际值）

  3. 从头消费历史消息（数据量大时可能较慢）：

    sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic device.1.data --from-beginning | grep '48985'
#                           └── 从最早的消息开始消费

    💡 可用临时 consumer group 避免影响现有消费进度：

    sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
    --topic device.1.data --from-beginning \
    --group debug-temp
#   └── 指定临时消费组名称，不影响生产消费进度

6.4 按阶段排查

  1. 消费 device.${TENANT_ID}.data，确认 source 容器已产出采集数据，详见第 3 章。
  2. 消费 original.${TENANT_ID}.data，确认 common 容器已转发处理后的数据，详见第 4 章。
  3. 消费 gongye.${TENANT_ID}.data 等业务域 Topic，确认 business 容器已筛选分发，详见第 5 章。

7 数据库排查

目的： 确认标准化数据已写入 PostgreSQL base_ods 原始数据层和 base_dwd 最终数据输出层。

7.1 base_ods 原始数据层

  1. 连接 viSCADA 数据库，查询 base_ods 原始数据表（表名中设备类标识替换为实际值，device_name 替换为实际设备名称），确认有新记录写入：

    SELECT *
FROM base_ods."viscada_v2_MEqZtns8wu"
WHERE device_name = '151101'
ORDER BY aligntime DESC
LIMIT 10;

    各部分说明：

    SELECT *
-- 查询所有字段
FROM base_ods."viscada_v2_MEqZtns8wu"
-- base_ods：原始数据层；"viscada_v2_MEqZtns8wu" 替换为实际设备类标识对应的表名
WHERE device_name = '151101'
-- 按设备名称过滤，替换为实际设备名称
ORDER BY aligntime DESC
-- 按采集时间降序排列，最新记录在前
LIMIT 10;
-- 只取 10 条，避免全表扫描

7.2 base_dwd 最终数据输出层

  1. 按设备 pid 查询历史数据表，确认有新记录写入：

    SELECT *
FROM base_dwd.substation_m_source_history
WHERE pid = 151101
ORDER BY aligntime DESC
LIMIT 10;

    各部分说明：

    SELECT *
-- 查询所有字段
FROM base_dwd.substation_m_source_history
-- base_dwd：最终数据输出层；substation_m_source_history 为历史数据表
WHERE pid = 151101
-- 按设备 pid 过滤，替换为实际设备 pid
ORDER BY aligntime DESC
-- 按采集时间降序排列，最新记录在前
LIMIT 10;
-- 只取 10 条，避免全表扫描

8 MQTT 推送排查

目的： 确认原始推送（conn/device/#）和标准化推送（conn/standardDevice/#）均有消息输出。

  1. 使用 MQTT 客户端（如 MQTTX）连接平台 MQTT Broker。
  2. 订阅 conn/device/#，确认可见原始点位值。
  3. 订阅 conn/standardDevice/#，确认可见标准化后的工程量值。

⚠️ 若 conn/device/# 有数据而 conn/standardDevice/# 无数据，问题出在 business 容器写出阶段，检查设备类或设备的**「标准化名称」**是否已填写。

9 常见报错分析

现象可能原因处理建议
容器状态异常或无法启动镜像拉取失败、资源不足或配置错误执行 kubectl -n prod describe pod <pod-name> 查看事件，定位启动失败原因
source 容器反复重启采集脚本语法错误或协议连接失败查看 source 容器日志，按 E! 过滤定位报错行；修复脚本后重新下发配置
source 日志无 metricStart 输出日志脚本未添加，或 namepass 与模板名不匹配按第 3.1 节检查日志脚本是否已添加，并核对 namepass 值与模板名是否一致
日志持续出现 E! 报错设备离线、协议参数有误或网络不通按设备类标识过滤日志，定位具体设备；核查**「通讯地址」**及网络连通性，参见第 2 章
source 容器正常但 Kafka device Topic 无数据采集脚本未生效或协议采集失败检查采集脚本是否已保存并下发；查看 source 容器日志确认有无 E! 报错
common 容器正常但 Kafka original Topic 无数据source → Kafka 链路中断先确认 device.${TENANT_ID}.data Topic 有数据；若无，问题在 source 阶段，参见第 3 章
business 容器正常但业务 Kafka Topic 无数据筛选条件不匹配或 business_id 配置错误确认 original.${TENANT_ID}.data Topic 有数据；检查 device.business_model_object 表中对应设备的 business_id 是否正确
ODS 表无新数据但 Kafka 有数据business 容器写出阶段失败查看 business 容器日志，按 E! 过滤定位写出阶段报错
DWD 表无新数据但 ODS 有数据历史拉链处理异常查看 business 容器日志，定位标准化阶段报错行
conn/standardDevice/# 无数据但 conn/device/# 有数据设备类或设备的**「标准化名称」**为空，或点位映射失败检查对应设备类或设备的**「标准化名称」**是否已填写；为空时数据不推送至标准化 Topic
conn/device/# 和 conn/standardDevice/# 均无数据source 或 common 阶段异常按数据流链路从上游逐层排查，参见第 1 章整体排查流程
Last updated on
808 connector 数据流链路说明文档01 Viscada 数据对接概念