跳到主要内容
版本:2.3.0-beta

Jdbc

JDBC sink connector

Description

Write data through jdbc

提示

Engine Supported and plugin name

  • Spark: Jdbc
  • Flink: Jdbc

Options

nametyperequireddefault value
driverstringyes-
urlstringyes-
userstringyes-
passwordstringyes-
dbTablestringyes-
saveModestringnoupdate
useSslstringnofalse
customUpdateStmtstringno-
duplicateIncsstringno-
showSqlstringnotrue

url [string]

The URL of the JDBC connection. Refer to a case: jdbc:mysql://localhost/dbName

user [string]

username

password [string]

user password

dbTable [string]

Sink table name, if the table does not exist, it will be created.

saveMode [string]

Storage mode, add mode update , perform data overwrite in a specified way when inserting data key conflicts

Basic mode, currently supports overwrite , append , ignore and error . For the specific meaning of each mode, see save-modes

useSsl [string]

Configure when saveMode is specified as update , whether to enable ssl, the default value is false

isolationLevel [string]

The transaction isolation level, which applies to current connection. The default value is READ_UNCOMMITTED

customUpdateStmt [string]

Configure when saveMode is specified as update , which is used to specify the update statement template for key conflicts. If customUpdateStmt is empty, the sql will auto-generate for all columns, else use the sql which refer to the usage of INSERT INTO table (...) values (...) ON DUPLICATE KEY UPDATE... of mysql , use placeholders or fixed values in values tips: the tableName of sql should be consistent with the dbTable.

duplicateIncs [string]

Configure when saveMode is specified as update , and when the specified key conflicts, the value is updated to the existing value plus the original value

showSql

Configure when saveMode is specified as update , whether to show sql

Examples

jdbc {
driver = "com.mysql.cj.jdbc.Driver",
saveMode = "update",
url = "jdbc:mysql://ip:3306/database",
user = "userName",
password = "***********",
dbTable = "tableName",
customUpdateStmt = "INSERT INTO table (column1, column2, created, modified, yn) values(?, ?, now(), now(), 1) ON DUPLICATE KEY UPDATE column1 = IFNULL(VALUES (column1), column1), column2 = IFNULL(VALUES (column2), column2)"
}

Insert data through JDBC

jdbc {
driver = "com.mysql.cj.jdbc.Driver",
saveMode = "update",
truncate = "true",
url = "jdbc:mysql://ip:3306/database",
user = "userName",
password = "***********",
dbTable = "tableName",
customUpdateStmt = "INSERT INTO tableName (column1, column2, created, modified, yn) values(?, ?, now(), now(), 1) ON DUPLICATE KEY UPDATE column1 = IFNULL(VALUES (column1), column1), column2 = IFNULL(VALUES (column2), column2)"
jdbc.connect_timeout = 10000
jdbc.socket_timeout = 10000
}

Timeout config