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

Datasource Supported Versions Driver Url Maven
SQL Server support version >= 2008 com.microsoft.sqlserver.jdbc.SQLServerDriver jdbc:sqlserver://localhost:1433 Download

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 Type SeaTunnel Data Type
BIT BOOLEAN
TINYINT
SMALLINT
SHORT
INTEGER INT
BIGINT LONG
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.)))
REAL FLOAT
FLOAT DOUBLE
CHAR
NCHAR
VARCHAR
NTEXT
NVARCHAR
TEXT
STRING
DATE LOCAL_DATE
TIME LOCAL_TIME
DATETIME
DATETIME2
SMALLDATETIME
DATETIMEOFFSET
LOCAL_DATE_TIME
TIMESTAMP
BINARY
VARBINARY
IMAGE
UNKNOWN
Not supported yet

Sink Options

Name Type Required Default Description
url String Yes - The URL of the JDBC connection. Refer to a case: jdbc:sqlserver://localhost:1433;databaseName=mydatabase
driver String Yes - The jdbc class name used to connect to the remote data source,
if you use sqlServer the value is com.microsoft.sqlserver.jdbc.SQLServerDriver.
user String No - Connection instance user name
password String No - Connection instance password
query String No - Use this sql write upstream input datas to database. e.g INSERT ...,query have the higher priority
database String No - 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.
table String No - 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_keys Array No - This option is used to support operations such as insert, delete, and update when automatically generate sql.
support_upsert_by_query_primary_key_exist Boolean No false Choose 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_sec Int No 30 The time in seconds to wait for the database operation used to validate the connection to complete.
max_retries Int No 0 The number of retries to submit failed (executeBatch)
batch_size Int No 1000 For 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_once Boolean No false Whether to enable exactly-once semantics, which will use Xa transactions. If on, you need to
set xa_data_source_class_name.
generate_sink_sql Boolean No false Generate sql statements based on the database table you want to write to
xa_data_source_class_name String No - 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_attempts Int No 3 The number of retries for transaction commit failures
transaction_timeout_sec Int No -1 The timeout after the transaction is opened, the default is -1 (never timeout). Note that setting the timeout may affect
exactly-once semantics
auto_commit Boolean No true Automatic transaction commit is enabled by default
common-options no - Sink plugin common parameters, please refer to Sink Common Options for details
enable_upsert Boolean No true Enable 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

  1. env {
  2. # You can set engine configuration here
  3. parallelism = 10
  4. }
  5. source {
  6. # This is a example source plugin **only for test and demonstrate the feature source plugin**
  7. Jdbc {
  8. driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
  9. url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  10. user = SA
  11. password = "Y.sa123456"
  12. query = "select * from column_type_test.dbo.full_types_jdbc"
  13. # Parallel sharding reads fields
  14. partition_column = "id"
  15. # Number of fragments
  16. partition_num = 10
  17. }
  18. # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
  19. # please go to https://seatunnel.apache.org/docs/connector-v2/source/Jdbc
  20. }
  21. transform {
  22. # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  23. # please go to https://seatunnel.apache.org/docs/transform-v2/sql
  24. }
  25. sink {
  26. Jdbc {
  27. driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
  28. url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  29. user = SA
  30. password = "Y.sa123456"
  31. 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( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )"
  32. } # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
  33. # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
  34. }

CDC(Change data capture) event

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

  1. Jdbc {
  2. source_table_name = "customers"
  3. driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
  4. url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  5. user = SA
  6. password = "Y.sa123456"
  7. generate_sink_sql = true
  8. database = "column_type_test"
  9. table = "dbo.full_types_sink"
  10. batch_size = 100
  11. primary_keys = ["id"]
  12. }

Exactly Once Sink

Transactional writes may be slower but more accurate to the data

  1. Jdbc {
  2. driver = com.microsoft.sqlserver.jdbc.SQLServerDriver
  3. url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  4. user = SA
  5. password = "Y.sa123456"
  6. 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( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )"
  7. is_exactly_once = "true"
  8. xa_data_source_class_name = "com.microsoft.sqlserver.jdbc.SQLServerXADataSource"
  9. } # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
  10. # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc