我正在使用 confluent REST API 代理调用 Kafka。我正在读取一个 CSV 文件,从那里的所有记录(大约 400 万条记录)中创建一个对象,并向 REST 代理发送请求。我不断收到OutOfMemory
异常。
确切的异常消息是:
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "kafka-producer-network-thread | producer-81"
我只有一个 REST 代理服务器实例,作为 docker 容器托管。环境变量设置为:
JAVA_OPTIONS=-Xmx1g
其他配置:
CPU - 1
Memory - 1024
它在崩溃之前处理了大约 1,00,000 个。我尝试将其扩展到 4 个实例,同时将 CPU 增加到 3,内存也增加到 2046 mb。然后它处理大约 5,00,000 条记录。
阅读 csv 后,我以 5k 条记录为单位调用 Kafka 端点。那是用Node写的。这是节点代码
fs.createReadStream(inputFile)
.pipe(parser({skip_lines_with_error: true}))
.on('data', (records) => {
country.push({ 'value' : {
country: records[0],
capital: records[1]
}
});
if (country.length > 5000) {
batch++;
callKafkaProxy(country).then((rec) => {
console.log(`'Batch done!'`);
}).catch((reason) => {
console.log(reason);
});
country = [];
}
})
.on('end', () => {
console.log('All done!');
});
function callKafkaProxy(records) {
const urlAndRequestOptions = {
url: 'http://kafka-rest-proxy.com/topics/test-topic',
headers: {
'content-type' : 'application/vnd.kafka.json.v2+json',
'Accept' : 'application/vnd.kafka.v2+json'
}
};
let recordsObject = {records: records};
//request here is a wrapper on the http package.
return request.post(urlAndRequestOptions, recordsObject);
我觉得我缺少一些配置,这些配置应该有助于解决这个问题,而不会增加实例数 > 1。
任何帮助将不胜感激。