跳到主要内容
版本：Next

SNMP

SNMPv2c SET Sink 连接器

描述​

SNMP Sink 为每一行输入向一个 SNMP Agent 发送一次同步 SNMPv2c SET 请求。 V1 范围仅包括 SET 操作，不发送 Trap 或 Inform，也不支持 SNMPv1 或 SNMPv3。

每一行需要提供数字 OID、字符串值和 SMI 值类型，对应的字段名可以配置。 默认映射会读取 SNMP Source 输出的 oid、value 和 value_type 字段；Sink 会忽略 agent、poll_time 等额外字段。

支持的引擎​

Spark
Flink
SeaTunnel Zeta

主要特性​

支持的数据源信息​

连接器使用 SNMP4J，支持通过 UDP 访问的 SNMPv2c Agent。

数据源支持版本依赖
SNMP AgentSNMPv2c下载

Sink 配置项​

名称类型是否必填默认值描述
hostString是-SNMP Agent 主机名或 IP 地址，不要包含协议或端口。
portInt否161SNMP Agent 的 UDP 端口。
communityString是-SNMPv2c community 凭证。连接器不会把该值写入日志或错误信息。
timeout_millisLong否5000每次 SET 请求尝试的超时时间，单位为毫秒。
retriesInt否1首次 SET 请求失败后的重试次数。0 表示只发送一次。
oid_fieldString否oid包含待设置数字 OID 的输入 STRING 字段。
value_fieldString否value包含待设置值的输入 STRING 字段。
value_type_fieldString否value_type包含 SMI 值类型的输入 STRING 字段。
common-options否-通用 Sink 配置项，包括 plugin_input。

工厂配置校验会在构造 Sink 之前检查字符串非空、端口范围、超时时间为正数以及重试次数非负。 三个映射字段必须存在于输入 Schema 中、类型必须为 STRING，并且不能指向同一个字段。Schema 错误会在创建任务时被拒绝。 空值以及空白的 OID 或值类型字段会在发送网络请求前被拒绝。值字段会根据其 SMI 类型进行校验；空的 OctetString 或 OctetStringHex 是有效值，文本 OctetString 的前后空白会被保留。

支持的 SMI 值类型​

value_type 不区分大小写，并忽略空白、_ 和 - 字符。 Sink 同时接受文档中的类型名和 SNMP Source 输出的 SNMP4J 语法字符串，包括 Counter、Gauge、OCTET STRING 和 OBJECT IDENTIFIER。

值类型可接受的值
Integer32 或 Integer有符号 32 位十进制整数。
UnsignedInteger32 或 UnsignedInteger0 到 4294967295 的十进制整数。
Counter32 或 Counter0 到 4294967295 的十进制整数。
Gauge32 或 Gauge0 到 4294967295 的十进制整数。
TimeTicks0 到 4294967295 的十进制百分之一秒计数，或 SNMP Source 使用的 SNMP4J 格式 [days, ]hours:mm:ss.hh。
Counter640 到 18446744073709551615 的十进制整数。
OctetString 或 OCTET STRING输入字符串表示的 UTF-8 文本。
OctetStringHex偶数个十六进制字符，例如 00ff10。
OID 或 OBJECT IDENTIFIER数字对象标识符，可以带前导点。
IpAddress点分 IPv4 地址。

OctetString 用于文本映射。如果需要逐字节保存二进制内容，请使用 OctetStringHex。

示例​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "snmp_updates"
    schema = {
      fields {
        oid = string
        value = string
        value_type = string
      }
    }
    rows = [
      {
        kind = INSERT
        fields = {
          oid = "1.3.6.1.2.1.1.5.0"
          value = "router-1"
          value_type = "OctetString"
        }
      }
    ]
  }
}

sink {
  SNMP {
    plugin_input = "snmp_updates"
    host = "192.0.2.10"
    port = 161
    community = "replace-with-your-community"
    timeout_millis = 3000
    retries = 1
  }
}

运行示例前请替换 community 占位值，并在已提交到源码的任务文件之外提供真实凭据。 记录解析后的任务配置时，community 会自动脱敏。无需为日志脱敏将其加入 shade.options； 该选项还参与配置的遮蔽和加密流程。

投递、失败和安全语义​

  • 一次 write 成功表示 Agent 已对该行返回成功的 SNMP 响应。
  • 所有配置尝试完成后仍超时，或 SNMP 响应包含非零错误状态时，Sink Task 会失败。
  • 一行在失败前可能阻塞约 timeout_millis * (retries + 1)。请确保该时间小于任务的 Checkpoint 超时时间。
  • SNMP4J 会重发超时请求。迟到的响应可能导致非幂等 OID 多次观察到同一次 SET。
  • Sink 没有事务提交协议或可恢复的 Writer 状态。引擎恢复后可能重复发送 SET，因此投递语义为至少一次。
  • 多个并行 Writer 可能乱序更新同一个 OID。如果更新顺序很重要，请使用并行度 1。
  • Sink 不会把 RowKind 解释为 CDC 操作。所有输入行（包括更新或删除类型）都会作为 SET 请求处理。
  • 请把 community 视为凭证，通过配置替换或其他密钥管理方式提供，不要把真实值提交到源码中的任务文件。
  • SNMPv2c 不提供传输加密或完整性保护，community 和 SET 负载会以明文发送。请仅在可信私有网络中使用，或通过 VPN 等受保护隧道传输。
  • Trap、Inform、SNMPv1 和 SNMPv3 不属于 V1 范围。

plugin_input 等配置请参阅通用 Sink 配置项。

Change Log
ChangeCommitVersion
[Feature][Connector-V2] Add SNMPv2c polling source connector-Next
[Feature][Connector-V2] Add SNMPv2c SET sink connector-Next