Блог

Я считываю случайные кортежи размером 500 МБ из Kafka producer continuous и в топологии storm вставляю их в MongoDB с помощью драйвера Java Mongo. Проблема в том, что я получаю действительно низкую пропускную способность в виде 4-5 кортежей в секунду.

Без вставки базы ДАННЫХ, если я напишу простой оператор печати, я получу пропускную способность в 684 кортежа в секунду. Я планирую запустить 1 миллион записей из Kafka и проверить пропускную способность с помощью mongo insert.

Я попытался настроить, используя параметры конфигурации setMaxSpoutPending , setMessageTimeoutSecs в kafkaconfig.

    final SpoutConfig kafkaConf = new SpoutConfig(zkrHosts, kafkaTopic, zkRoot, clientId);
    kafkaConf.ignoreZkOffsets=false;
    kafkaConf.useStartOffsetTimeIfOffsetOutOfRange=true;
    kafkaConf.startOffsetTime=kafka.api.OffsetRequest.LatestTime();
    kafkaConf.stateUpdateIntervalMs=2000;
    kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());
    final TopologyBuilder topologyBuilder = new TopologyBuilder();
    topologyBuilder.setSpout("kafka-spout", new KafkaSpout(kafkaConf), 1);
    topologyBuilder.setBolt("print-messages", new MyKafkaBolt()).shuffleGrouping("kafka-spout");
     Config conf = new Config();
     conf.setDebug(true);
     conf.setMaxSpoutPending(1000);
     conf.setMessageTimeoutSecs(30);
  

Выполнить метод bolt

       JSONObject jObj = new JSONObject();
    jObj.put("key", input.getString(0));

        if (null !=jObj amp;amp; jObj.size() > 0 ) {
            final DBCollection quoteCollection = dbConnect.getConnection().getCollection("stormPoc");
            if (quoteCollection != null) {
                BasicDBObject dbObject = new BasicDBObject();
                dbObject.putAll(jObj);
                quoteCollection.insert(dbObject);
            //  logger.info("inserted in Collection !!!");
            } else {
                logger.info("Error while inserting data in DB!!!");
            }
            collector.ack(input);
  

Существует storm-mongodb модуль для интеграции с Mongo. Разве это не выполняет свою работу? https://github.com/apache/storm/tree/b07413670fa62fec077c92cb78fc711c3bda820c/external/storm-mongodb

Вы не должны использовать storm-kafka для интеграции Kafka, она устарела. Используйте storm-kafka-client вместо этого.

Настройка conf.setDebug(true) повлияет на вашу обработку, поскольку Storm будет регистрировать довольно большое количество текста для каждого кортежа.