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

【问题标题】：Processing huge records gives OutOfMemoryException - Kafka REST proxy处理大量记录会产生 OutOfMemoryException - Kafka REST 代理
【发布时间】：2019-05-18 18:47:15
【问题描述】：

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

任何帮助将不胜感激。

【问题讨论】：

你能给出你的代码的sn-p你是如何批量记录到Kafka的吗？您似乎没有使用节点流或以错误的异步方式进行操作。
刚刚添加@ŽilvinasJocius，您可以查看。
Kafka REST 客户端真的适合你的需求吗？如果客户端没有像 http 那样的额外层但直接使用 tcp/ip 进行通信呢？如果您能找到支持流的客户端，您的任务将执行得更快而不会出现此类问题。
其余客户端现在在生产中的很多地方都在使用。但是，目前，它产生的消息要低几个数量级。直接生产到 kafka 是可能的，但不方便并且禁止代码重用（因为它已经用于其他应用程序）。
你明白我在下面的回答中试图解释的问题吗？

标签： node.js apache-kafka kafka-producer-api kafka-rest

【解决方案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 = [];
     }

你的 callKafkaProxy 是异步的，这就是为什么你的国家数组总是被填充，不管 callKafkaProxy 函数的结果如何。 Country 数组不断填充并不断发出请求。您可以通过批处理++之后的控制台日志记录来确保。您会看到您正在发起大量请求，而 Kafka 的响应速度会比您发出请求的速度慢得多。

解决方案：

创建可写流。
将数据从您的解析器传送到它。 input.pipe(parser).pipe(yourJustCreatedKafkaWritableStream)
当您准备好接收其他记录时，让您的可写流将国家/地区推送到数组和回调。当您到达边缘时（如果 countries.length > 5000），则向 kafka 发出请求并等待响应，然后才进行回调。这样，您的流将是自适应的。您应该阅读有关节点流及其功能的更多信息。但请记住，权力越大，责任越大，在这种情况下，您必须仔细设计代码以避免此类内存泄漏。

【讨论】：

我了解#2。它可能需要是同步调用或调用之间有一些延迟。我仍然没有关注#1。背压是什么意思，.pipe(kafka) 是什么？
不要使用随机延迟。问题是 .on('data') 事件侦听器不会等到您执行某些操作并且它会不断发出。这会导致您的代码始终将数据推送到 kafka。您的检查 if(country.length > ...) 是多余的，因为 'data' 事件不断将记录添加到您的数组中。因此，如果您对其进行调试，您将看到无论响应如何，您都将记录推送到 kafka。您应该以流的方式进行。读取文件->解析->推送到kafka。 Kafka 是这里的瓶颈，这就是为什么你必须创建可写流来告诉其他人工作慢的原因。
谢谢。我明白你在说什么。我需要看看如何使用流。 csv 解析器可用作流。我会尝试一下，看看是否能解决我的问题。
感谢您为我指明了正确的方向。我已经使用fs.createReadStream 让它工作了。我将发布带有详细信息的答案，因此任何转向此问题的人都会发现该实现很有用。

【解决方案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();
                                });
                            }

【讨论】：