Sink JDBC how to control the batch size.

[isMrH]

I want to control the number of processes that sink JDBC writes to tidb and the batch size.
How should I configure it?

The configuration file is as follows:

sink {
  jdbc {
    driver = "com.mysql.cj.jdbc.Driver"
    saveMode = "update"
    url = "jdbc:mysql://xx.xx.xx.xx:4000/xxx?&useConfigs=maxPerformance&useServerPrepStmts=true&prepStmtCacheSqlLimit=2048&prepStmtCacheSize=256&rewriteBatchedStatements=true&allowMultiQueries=true"
    user = "tidb"
    password = "tidb"
    dbTable = "table"
    isolationLevel = "NONE"
    source_table_name = "source_table_name"
  }
}

error:

java.sql.BatchUpdateException: Out Of Memory Quota!

start:

./bin/start-seatunnel-spark.sh 

[ashulin]

add options, prefix is 'jdbc.', eg, 'jdbc.batchsize':

1. batchsize
2. numPartitions
   See spark v2.4 jdbc for more options and comments

[isMrH]

According to the official documentation, I only see this parameter in source.
Are you sure you can use it in Sink?

[isMrH]

@ashulin
I don't want to set the partitionColumn.
So I only added the following two configurations, but nothing changed in the process of inserting the data.
tidb process count is still high.
bath_size is still large.

jdbc.numPartitions = 8
jdbc.batchsize = 200

error

Job aborted due to stage failure: Task 177 in stage 4.0 failed 4 times, most recent failure: Lost task 177.3 in stage 4.0 (TID 4100, hdfs-14.qd.link-x.host, executor 202): java.sql.BatchUpdateException: Out Of Memory Quota![conn_id=249827]

[isMrH]

@ashulin Thank you for your answer, but my problem is still unresolved.
My Source is using hive for reading.
I want to write to tidb via Sink-spark-jdbc.
The jdbc.* configuration item is only described in the official documentation Source.
I added partitionColumn,lowerBound,upperBound,numPartitions to Sink-tidb
No change occurred.

env {
  spark.app.name = "st_export_test"
  spark.executor.instances = 20
  spark.executor.cores = 4
  spark.executor.memory = "8g"
  spark.sql.catalogImplementation = "hive"
}

source {
  hive {
    pre_sql = """
    select * from table
    """
    result_table_name = "table"
  }
}

transform {
}

sink {
  jdbc {
    driver = "com.mysql.cj.jdbc.Driver"
    saveMode = "update"
    url = "jdbc:mysql://10.32.xx.xx:4000/test?&useConfigs=maxPerformance&useServerPrepStmts=true&prepStmtCacheSqlLimit=2048&prepStmtCacheSize=256&rewriteBatchedStatements=true&allowMultiQueries=true"
    user = "tidb"
    password = "tidb"
    dbTable = "table"
    isolationLevel = "NONE"
    jdbc.partitionColumn = "id"
    jdbc.lowerBound = 1
    jdbc.upperBound = 1000
    jdbc.numPartitions = 10
    source_table_name = "table"
  }
}

[ashulin]

Sorry, I made a mistake.

[isMrH]

What should I do next?

[ashulin]

Sorry, that's a issue; I have looked at the Spark JDBC Sink code and it does not resolve the extra jdbc options;

[isMrH]

I see, I'll try other avenues. I hope this problem can be solved in the future.Thanks for your help!