alibaba / alibaba/DataX

clickhouse 当writeBuffer根据batchsize设置过大,容易产生gc

Open
#1,291 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
17.4k
Forks
5.7k
PR merge metrics
No merged PRs in 30d

Description

版本分支:master(3.0)
场景:odps的tunnel读取,写入clickhouse
原因:由于clickhouse插入要求批量,至少在1000条每批,根据建议能50M~200M/s,因此我们设置batchsize=50w, 直接轮询插入clickhouse的本地_local表, datax堆设置-Xms100g -Xmx100g,设置channel不限速,开始过段时间就出现gc
gc分析:
代码
com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter.Task#startWriteWithConnection 和com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter.Task#doBatchInsert代码
public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) {
this.taskPluginCollector = taskPluginCollector;

// 用于写入数据的时候的类型根据目的表字段类型转换
this.resultSetMetaData = DBUtil.getColumnMetaData(connection,
this.table, StringUtils.join(this.columns, ","));
// 写数据库的SQL语句
calcWriteRecordSql();

List writeBuffer = new ArrayList(this.batchSize);
int bufferBytes = 0;
try {
Record record;
while ((record = recordReceiver.getFromReader()) != null) {
if (record.getColumnNumber() != this.columnNumber) {
// 源头读取字段列数与目的表字段写入列数不相等,直接报错
throw DataXException
.asDataXException(
DBUtilErrorCode.CONF_ERROR,
String.format(
"列配置信息有错误. 因为您配置的任务中,源头读取字段数:%s 与 目的表要写入的字段数:%s 不相等. 请检查您的配置并作出修改.",
record.getColumnNumber(),
this.columnNumber));
}

writeBuffer.add(record);
bufferBytes += record.getMemorySize();

if (writeBuffer.size() >= batchSize || bufferBytes >= batchByteSize) {
doBatchInsert(connection, writeBuffer);
writeBuffer.clear();
bufferBytes = 0;
}
}
if (!writeBuffer.isEmpty()) {
doBatchInsert(connection, writeBuffer);
writeBuffer.clear();
bufferBytes = 0;
}
} catch (Exception e) {
throw DataXException.asDataXException(
DBUtilErrorCode.WRITE_DATA_ERROR, e);
} finally {
writeBuffer.clear();
bufferBytes = 0;
DBUtil.closeDBResources(null, null, connection);
}
}

protected void doBatchInsert(Connection connection, List buffer)
throws SQLException {
PreparedStatement preparedStatement = null;
try {
connection.setAutoCommit(false);
preparedStatement = connection
.prepareStatement(this.writeRecordSql);

for (Record record : buffer) {
preparedStatement = fillPreparedStatement(
preparedStatement, record);
preparedStatement.addBatch();
}
preparedStatement.executeBatch();
connection.commit();
} catch (SQLException e) {
LOG.warn("回滚此次写入, 采用每次写入一行方式提交. 因为:" + e.getMessage());
connection.rollback();
doOneInsert(connection, buffer);
} catch (Exception e) {
throw DataXException.asDataXException(
DBUtilErrorCode.WRITE_DATA_ERROR, e);
} finally {
DBUtil.closeDBResources(preparedStatement, null);
}
}

根据java_pidxx.hprof分析writeBuffer 和 preparedStatement 同时存在,占用两倍内存空间,慢慢根据jvm垃圾回收出现gc,速度也下降下来
尝试修改:去掉writeBuffer ,直接操作preparedStatement.addBatch();,达到batchsize,就preparedStatement.executeBatch(),
看jvm内存减少接近一半,速度变稳定了,速度提升10倍以上,就没有发生gc

提问:这种修改是有效?或者其他有没有其他方法?jvm 调参试过,尝试把年轻代-XX:NewSize调占比重很大,达到3/4之类

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by reading com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter.Task#startWriteWithConnection and #doBatchInsert, then review the reported java_pidxx.hprof findings. Compare the existing writeBuffer and PreparedStatement batching behavior, and validate the chosen approach with a representative ClickHouse write while checking heap usage, GC activity, throughput, and fallback behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
clickhouse, java
Domain
backend, databases, performance
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.