跳到主要内容
版本：3.0.0

OpenMldb

OpenMldb 源连接器

支持的引擎​

Spark
Flink
SeaTunnel Zeta

描述​

用于从 OpenMLDB 读取数据。连接器会执行配置的 SQL 语句并把结果转换为 SeaTunnel 记录，同时支持 单机版和集群版两种部署模式。

关键特性​

数据类型映射​

OpenMLDB 类型会按照所配置 sql 语句的结果集映射为 SeaTunnel 类型。SeaTunnel 不原生支持的类型会直接 导致读取失败，并抛出 UNSUPPORTED_DATA_TYPE 错误。

OpenMLDB 数据类型SeaTunnel 数据类型
boolboolean
smallintsmallint
intint
bigintbigint
float / doublefloat / double
string / varcharstring
datedate
timestamptimestamp

选项​

名称类型必需默认值描述
cluster_modeboolean是-是否以 OpenMLDB 集群模式连接。false 表示单机模式，true 表示集群模式。
sqlstring是-用于读取数据的 SQL 语句，列名和类型按结果集定义。
databasestring是-要连接的 OpenMLDB 数据库名称。
hoststring否-当 cluster_mode 为 false 时必填，OpenMLDB 单机版主机地址。
portint否-当 cluster_mode 为 false 时必填，OpenMLDB 单机版端口。
zk_hoststring否-当 cluster_mode 为 true 时必填，OpenMLDB 集群对应的 ZooKeeper 地址列表。
zk_pathstring否-当 cluster_mode 为 true 时必填，OpenMLDB 集群在 ZooKeeper 上的路径，例如 /openmldb。
session_timeoutint否10000OpenMLDB 会话超时时间，单位毫秒。
request_timeoutint否60000OpenMLDB 请求超时时间，单位毫秒。
common-options否-源插件通用参数，详见 Source 常见选项。

cluster_mode [boolean]​

是否以 OpenMLDB 集群模式连接。为 false 时配置 host 和 port；为 true 时配置 zk_host 和 zk_path。

sql [string]​

无论使用单机模式还是集群模式，必填项 sql 都不能为空字符串或仅包含空白字符。

针对 OpenMLDB 执行的 SQL 语句，结果集的列会成为连接器输出行的字段。

database [string]​

要连接的 OpenMLDB 数据库名称，配置的数据库必须在目标 OpenMLDB 实例上存在。

host [string]​

OpenMLDB 主机，仅在 cluster_mode 为 false（单机模式）下使用。

port [int]​

OpenMLDB 端口，仅在 cluster_mode 为 false（单机模式）下使用。

zk_host [string]​

OpenMLDB 集群对应的 ZooKeeper 地址列表，例如 zk-1:2181,zk-2:2181,zk-3:2181，仅在 cluster_mode 为 true 时使用。

zk_path [string]​

OpenMLDB 集群在 ZooKeeper 上的路径，例如 /openmldb，仅在 cluster_mode 为 true 时使用。

session_timeout [int]​

OpenMLDB 会话超时时间，单位毫秒，默认 10000（10 秒）。

request_timeout [int]​

OpenMLDB 请求超时时间，单位毫秒，默认 60000（60 秒）。

common options​

源插件通用参数，详见 Source 常见选项。

任务示例​

单机模式​

source {
  OpenMldb {
    host = "172.17.0.2"
    port = 6527
    sql = "select * from demo_table1"
    database = "demo_db"
    cluster_mode = false
  }
}

集群模式​

source {
  OpenMldb {
    zk_host = "zk-1:2181,zk-2:2181,zk-3:2181"
    zk_path = "/openmldb"
    sql = "select * from demo_table1"
    database = "demo_db"
    cluster_mode = true
  }
}

配合下游接收器​

从 OpenMLDB 读取数据并通过 Console 接收器打印的典型端到端作业。

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

source {
  OpenMldb {
    host = "172.17.0.2"
    port = 6527
    sql = "select id, name from demo_table1"
    database = "demo_db"
    cluster_mode = false
  }
}

sink {
  Console {
  }
}

变更日志​

Change Log
ChangeCommitVersion
[Improve][Connector-V2] Validate OpenMLDB SQL (#12181)https://github.com/apache/seatunnel/commit/30474f8633.0.0
[improve] openmldb options (#9166)https://github.com/apache/seatunnel/commit/d324fc59a42.3.11
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Feature][Conenctor-V2] Add openmldb source connector (#3313)https://github.com/apache/seatunnel/commit/e68ecf7bef2.3.0