Jdbc

JDBC source connector

Description

Read external data source data through JDBC

tip

Engine Supported and plugin name

• Spark: Jdbc
• Flink: Jdbc

Options

Spark

name	type	required	default value
driver	string	yes	-
jdbc.*	string	no
password	string	yes	-
table	string	yes	-
url	string	yes	-
user	string	yes	-
common-options	string	yes	-

Flink

name	type	required	default value
driver	string	yes	-
url	string	yes	-
username	string	yes	-
password	string	no	-
query	string	yes	-
fetch_size	int	no	-
partition_column	string	no	-
partition_upper_bound	long	no	-
partition_lower_bound	long	no	-
common-options	string	no	-
parallelism	int	no	-

driver [string]

The jdbc class name used to connect to the remote data source, if you use MySQL the value is com.mysql.cj.jdbc.Driver.

Warn: for license compliance, you have to provide MySQL JDBC driver yourself, e.g. copy mysql-connector-java-xxx.jar to $FLINK_HOME/lib for Standalone.

password [string]

password

url [string]

The URL of the JDBC connection. Refer to a case: jdbc:postgresql://localhost/test

Spark

jdbc [string]

In addition to the parameters that must be specified above, users can also specify multiple optional parameters, which cover all the parameters provided by Spark JDBC.

The way to specify parameters is to add the prefix jdbc. to the original parameter name. For example, the way to specify fetchsize is: jdbc.fetchsize = 50000 . If these non-essential parameters are not specified, they will use the default values given by Spark JDBC.

user [string]

username

table [string]

table name

Flink

username [string]

username

query [string]

Query statement

fetch_size [int]

fetch size

parallelism [int]

The parallelism of an individual operator, for JdbcSource.

partition_column [string]

The column name for parallelism's partition, only support numeric type.

partition_upper_bound [long]

The partition_column max value for scan, if not set SeaTunnel will query database get max value.

partition_lower_bound [long]

The partition_column min value for scan, if not set SeaTunnel will query database get min value.

common options [string]

Source plugin common parameters, please refer to Source Plugin for details

Example

Spark

jdbc {
    driver = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/info"
    table = "access"
    result_table_name = "access_log"
    user = "username"
    password = "password"
}

Read MySQL data through JDBC

jdbc {
    driver = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/info"
    table = "access"
    result_table_name = "access_log"
    user = "username"
    password = "password"
    jdbc.partitionColumn = "item_id"
    jdbc.numPartitions = "10"
    jdbc.lowerBound = 0
    jdbc.upperBound = 100
}

Divide partitions based on specified fields

jdbc {
    driver = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/info"
    table = "access"
    result_table_name = "access_log"
    user = "username"
    password = "password"
    
    jdbc.connect_timeout = 10000
    jdbc.socket_timeout = 10000
}

Timeout config

Flink

JdbcSource {
    driver = com.mysql.cj.jdbc.Driver
    url = "jdbc:mysql://localhost/test"
    username = root
    query = "select * from test"
}