处理大量记录会产生OutOfMemoryException - Kafka REST代理

Question

我正在使用汇合的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。

任何帮助将不胜感激。

Answer 1

.on('data', () => {}); ... 

1.它不处理背压。创建可写流，它将处理您的批处理过程。然后只需使用管道。

inputStream
    .pipe(parser)
    .pipe(kafka)

然后分析这些线：

if (country.length > 5000) {
        batch++;
        callKafkaProxy(country).then((rec) => {
            console.log(`'Batch done!'`);
        ).catch((reason) => {
            console.log(reason);
        });
        country = [];
     }

2. 你的callKafkaProxy是异步的，这就是你的country数组总是被填充的原因，无论callKafkaProxy函数的结果如何。国家阵列不断填写并不断提出请求。您可以通过批处理++之后的控制台记录来确保。您将看到您正在发起大量请求，而Kafka的响应速度比您提出请求要慢得多。

解：

1. 创建可写流。
2. 从解析器管道数据到它。 input.pipe（解析器）.pipe（yourJustCreatedKafkaWritableStream）
3. 当您准备好接收其他记录时，让您的可写流将国家/地区推送到阵列和回调。当你到达边缘时（如果countries.length> 5000）然后向kafka发出请求并等待响应，然后才给出回调。通过这种方式，您的流将是自适应的。您应该阅读有关节点流及其功能的更多信息。但请记住，强大的功能带来了巨大的责任，在这种情况下，您必须仔细设计代码以避免此类内存泄漏。

Answer 2

在Zilvinas的回答的帮助下，我理解了如何利用流来批量发送数据。这是一个解决方案：

var stream = fs.createReadStream(file)
                        .pipe(es.split())
                        .pipe(es.mapSync(function (line) {

                            if (line.length) {
                                //read your line and create a record message
                            }

                            //put 5000 in a config constant
                            if (records.length === 5000) {
                                stream.pause();
                                logger.debug(`Got ${records.length} messages. Pushing to Kafka...`);
                                postChunkToKafka(records).then((response) => {     
                                  records = [];
                                  stream.resume();
                                });
                            }