跳到主要内容
版本:Next

SQL Server

JDBC SQL Server Sink Connector

Support SQL Server Version

  • server:2008 (Or later version for information only)

Support Those engines

Spark
Flink
Seatunnel Zeta

Description

Write data through jdbc. Support Batch mode and Streaming mode, support concurrent writing, support exactly-once semantics (using XA transaction guarantee).

Using Dependency

  1. You need to ensure that the jdbc driver jar package has been placed in directory ${SEATUNNEL_HOME}/plugins/.

For SeaTunnel Zeta Engine

  1. You need to ensure that the jdbc driver jar package has been placed in directory ${SEATUNNEL_HOME}/lib/.

Key Features

Use Xa transactions to ensure exactly-once. So only support exactly-once for the database which is support Xa transactions. You can set is_exactly_once=true to enable it.

Supported DataSource Info

DatasourceSupported VersionsDriverUrlMaven
SQL Serversupport version >= 2008com.microsoft.sqlserver.jdbc.SQLServerDriverjdbc:sqlserver://localhost:1433Download

Database dependency

Please download the support list corresponding to 'Maven' and copy it to the '$SEATNUNNEL_HOME/plugins/jdbc/lib/' working directory
For example SQL Server datasource: cp mssql-jdbc-xxx.jar $SEATNUNNEL_HOME/plugins/jdbc/lib/

Data Type Mapping

SQLserver Data TypeSeaTunnel Data Type
BITBOOLEAN
TINYINT
SMALLINT
SHORT
INTEGERINT
BIGINTLONG
DECIMAL
NUMERIC
MONEY
SMALLMONEY
DECIMAL((Get the designated column's specified column size)+1,
(Gets the designated column's number of digits to right of the
decimal point.)))
REALFLOAT
FLOATDOUBLE
CHAR
NCHAR
VARCHAR
NTEXT
NVARCHAR
TEXT
STRING
DATELOCAL_DATE
TIMELOCAL_TIME
DATETIME
DATETIME2
SMALLDATETIME
DATETIMEOFFSET
LOCAL_DATE_TIME
TIMESTAMP
BINARY
VARBINARY
IMAGE
UNKNOWN
Not supported yet

Sink Options

NameTypeRequiredDefaultDescription
urlStringYes-The URL of the JDBC connection. Refer to a case: jdbc:sqlserver://localhost:1433;databaseName=mydatabase
driverStringYes-The jdbc class name used to connect to the remote data source,
if you use sqlServer the value is com.microsoft.sqlserver.jdbc.SQLServerDriver.
userStringNo-Connection instance user name
passwordStringNo-Connection instance password
queryStringNo-Use this sql write upstream input datas to database. e.g INSERT ...,query have the higher priority
databaseStringNo-Use this database and table-name auto-generate sql and receive upstream input datas write to database.
This option is mutually exclusive with query and has a higher priority.
tableStringNo-Use database and this table-name auto-generate sql and receive upstream input datas write to database.
This option is mutually exclusive with query and has a higher priority.
primary_keysArrayNo-This option is used to support operations such as insert, delete, and update when automatically generate sql.
support_upsert_by_query_primary_key_existBooleanNofalseChoose to use INSERT sql, UPDATE sql to process update events(INSERT, UPDATE_AFTER) based on query primary key exists. This configuration is only used when database unsupport upsert syntax. Note: that this method has low performance
connection_check_timeout_secIntNo30The time in seconds to wait for the database operation used to validate the connection to complete.
max_retriesIntNo0The number of retries to submit failed (executeBatch)
batch_sizeIntNo1000For batch writing, when the number of buffered records reaches the number of batch_size or the time reaches checkpoint.interval
, the data will be flushed into the database
is_exactly_onceBooleanNofalseWhether to enable exactly-once semantics, which will use Xa transactions. If on, you need to
set xa_data_source_class_name.
generate_sink_sqlBooleanNofalseGenerate sql statements based on the database table you want to write to
xa_data_source_class_nameStringNo-The xa data source class name of the database Driver, for example, SqlServer is com.microsoft.sqlserver.jdbc.SQLServerXADataSource, and
please refer to appendix for other data sources
max_commit_attemptsIntNo3The number of retries for transaction commit failures
transaction_timeout_secIntNo-1The timeout after the transaction is opened, the default is -1 (never timeout). Note that setting the timeout may affect
exactly-once semantics
auto_commitBooleanNotrueAutomatic transaction commit is enabled by default
common-optionsno-Sink plugin common parameters, please refer to Sink Common Options for details
enable_upsertBooleanNotrueEnable upsert by primary_keys exist, If the task has no key duplicate data, setting this parameter to false can speed up data import

tips

If partition_column is not set, it will run in single concurrency, and if partition_column is set, it will be executed in parallel according to the concurrency of tasks.

Task Example

simple:

This is one that reads Sqlserver data and inserts it directly into another table

env {
# You can set engine configuration here
parallelism = 10
}

source {
# This is a example source plugin **only for test and demonstrate the feature source plugin**
Jdbc {
driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
user = SA
password = "Y.sa123456"
query = "select * from column_type_test.dbo.full_types_jdbc"
# Parallel sharding reads fields
partition_column = "id"
# Number of fragments
partition_num = 10

}
# If you would like to get more information about how to configure seatunnel and see full list of source plugins,
# please go to https://seatunnel.apache.org/docs/connector-v2/source/Jdbc
}

transform {

# If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
# please go to https://seatunnel.apache.org/docs/transform-v2/sql
}

sink {
Jdbc {
driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
user = SA
password = "Y.sa123456"
query = "insert into full_types_jdbc_sink( id, val_char, val_varchar, val_text, val_nchar, val_nvarchar, val_ntext, val_decimal, val_numeric, val_float, val_real, val_smallmoney, val_money, val_bit, val_tinyint, val_smallint, val_int, val_bigint, val_date, val_time, val_datetime2, val_datetime, val_smalldatetime ) values( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )"

} # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
# please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
}

CDC(Change data capture) event

CDC change data is also supported by us In this case, you need config database, table and primary_keys.

Jdbc {
source_table_name = "customers"
driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
user = SA
password = "Y.sa123456"
generate_sink_sql = true
database = "column_type_test"
table = "dbo.full_types_sink"
batch_size = 100
primary_keys = ["id"]
}

Exactly Once Sink

Transactional writes may be slower but more accurate to the data

  Jdbc {
driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
user = SA
password = "Y.sa123456"
query = "insert into full_types_jdbc_sink( id, val_char, val_varchar, val_text, val_nchar, val_nvarchar, val_ntext, val_decimal, val_numeric, val_float, val_real, val_smallmoney, val_money, val_bit, val_tinyint, val_smallint, val_int, val_bigint, val_date, val_time, val_datetime2, val_datetime, val_smalldatetime ) values( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )"
is_exactly_once = "true"

xa_data_source_class_name = "com.microsoft.sqlserver.jdbc.SQLServerXADataSource"

} # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
# please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc