idea maven 依赖

<dependency>
<groupId>ru.yandex.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.2.4</version>
</dependency>

val clickhouseUri = "jdbc:clickhouse://host:8123/aso?user=root&password=XXX" 

val table =""

processed.addSink(
JdbcSink.sink(
"insert into $table values(?,?)",
CKSinkBuilder(),
JdbcExecutionOptions.Builder()
.withBatchIntervalMs(20000)  
.withBatchSize(20000).build(),  //20s一批或20000条数据，那个先达到都执行sink
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl(clickhouseUri)
.withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
.build()
)
)
 

class CKSinkBuilder : JdbcStatementBuilder<DemoData> {
override fun accept(ps: PreparedStatement, v: DemoData) {
ps.setLong(1, v.ts!!)
ps.setSTring(2, v.data!!)

//还有其他字段接着set ps.setSTring(3, v.XXX!!)
}
}

class DemoData{
var ts: Long? = null
var data: String? = null
}