跳到主要内容
版本:Next

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_modefalse 时必填,OpenMLDB 单机版主机地址。
portint-cluster_modefalse 时必填,OpenMLDB 单机版端口。
zk_hoststring-cluster_modetrue 时必填,OpenMLDB 集群对应的 ZooKeeper 地址列表。
zk_pathstring-cluster_modetrue 时必填,OpenMLDB 集群在 ZooKeeper 上的路径,例如 /openmldb
session_timeoutint10000OpenMLDB 会话超时时间,单位毫秒。
request_timeoutint60000OpenMLDB 请求超时时间,单位毫秒。
common-options-源插件通用参数,详见 Source 常见选项

cluster_mode [boolean]

是否以 OpenMLDB 集群模式连接。为 false 时配置 hostport;为 true 时配置 zk_hostzk_path

sql [string]

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

database [string]

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

host [string]

OpenMLDB 主机,仅在 cluster_modefalse(单机模式)下使用。

port [int]

OpenMLDB 端口,仅在 cluster_modefalse(单机模式)下使用。

zk_host [string]

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

zk_path [string]

OpenMLDB 集群在 ZooKeeper 上的路径,例如 /openmldb,仅在 cluster_modetrue 时使用。

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] 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