Как контролировать общий размер данных, производимых несколькими партиями, с помощью Spark Kafka Producer в Java - PullRequest
Новая
       162

Как контролировать общий размер данных, производимых несколькими партиями, с помощью Spark Kafka Producer в Java

1 голос
/ 10 июля 2020

У меня есть Spark Kafka Producer (написано на Java), который запускается партиями на ежедневной основе. Я хочу контролировать размер данных, производимых за день.

Допустим, я хочу произвести общий объем данных размером 100 ГБ (все пакеты вместе) за день несколькими партиями, а затем я хочу остановить обработать. Как я могу рассчитать и сохранить размер (в байтах) данных, производимых в каждом пакете, чтобы я мог остановить процесс, как только общее количество достигнет отметки 100 ГБ?

Ниже приведены API-интерфейсы, которые я использую у производителя: 

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;

Производитель выглядит так: 

public void produceData(String key, String val) {
    
    if (null == kafkaProducer) {
        
        kafkaProducer = new KafkaProducer<String, String>(<Kafka Parameters>);
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                kafkaProducer.close();
            }
        });
        
    }
    
    if (key != null) {
        
        kafkaProducer.send(new ProducerRecord(topic, key, val), new ProducerCallback(key, val) {
            public void flush() {
                if (null != kafkaProducer) {
                    kafkaProducer.flush();
                }
            }
        });
        
    }
}
...