Приложение-загрузчик Cassandra hector не хватает памяти

Question

Это простое приложение берет файл с разделителями-запятыми с заголовками и помещает их в Cassandra. Это работает для небольшого файла, однако память просто увеличивается, пока исключение из нехватки памяти не убьет его.

Что мне не хватает?

package com.company;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

public class QuickLoad {
    public static Keyspace keyspace = null;
    public static void main(String[] args) {
        File file = new File(args[0]);
        String keyspaceName = args[1];
        String columnFamilyName = args[2];
        BufferedReader reader = null;
        try {
            keyspace = GetKeyspace(keyspaceName);
            reader = new BufferedReader(new FileReader(file));
            String fileLine = null;
            String[] headers = null;
            String[] fields = null;
            boolean headerLine = true;

            while ((fileLine = reader.readLine()) != null) {
                if (headerLine){
                    headerLine = false;
                    headers = fileLine.substring(1, fileLine.length()-1).split("\",\"");
                } else {
                    fields = fileLine.substring(1, fileLine.length()-1).split("\",\"");
                    CassandraSave(keyspace, columnFamilyName, headers, fields);
                }
            }
        }
        catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (reader != null) {
                    reader.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        System.exit(0);
    }

    public static void CassandraSave(Keyspace keyspace, String columnFamily, String[] headers, String[] columns) 
    {
        try 
        {
            Mutator mutator = HFactory.createMutator(keyspace, StringSerializer.get());
            for (int i = 1; i < headers.length-1; i++)
            {
                if ((columns[i] != null) || (!columns[i].equals("null"))) {
                    if (columns[i].length() > 0) {
                        HColumn<String, String> col = HFactory.createStringColumn(headers[i], columns[i]);
                        mutator.insert(columns[1], columnFamily, col);
                    }
                }
            }
            mutator.execute();
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    public static Keyspace GetKeyspace(String keyspaceName)
    {
        String serverAddress = "localhost:9160";
        Cluster cluster = HFactory.getOrCreateCluster("My Cluster", serverAddress);
        Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
        return keyspace;
    }

}

Answer 1

Один экземпляр "com.ecyrd.speed4j.log.PeriodicalLog", загруженный "sun.misc.Launcher$AppClassLoader @ 0x899902f8", занимает 127 293 432 (99,62%) байта. Ключевые слова com.ecyrd.speed4j.log.PeriodicalLog sun.misc.Launcher$AppClassLoader @ 0x899902f8

Похоже, вы используете более старую версию hector и сталкиваетесь с ошибкой с утечкой памяти в speed4j. Если вы обновитесь до hector 0.8.0-2, это должно быть исправлено.

Стоит отметить, что speed4j отключен по умолчанию в 0.8.0-2, если вы хотите включить его, посмотрите эту ветку.

Answer 2

Я мог бы видеть это как проблему, если один из ваших "столбцов" во входном файле был больше, чем выделенная куча. Вы можете исправить это, установив верхнюю границу размера вашей мутации, так как ваша функция CassandraSave делает только 100 или около того мутаций за одну операцию.

Answer 3

Я вижу две вещи - это однопоточный и размер партии довольно маленький.

Добавьте внешний цикл, чтобы собрать вставки в мутаторе с размером партии около 500 строк, чтобы начать и посмотреть, как это происходит. Вот пример вставки мутанта, который я использую для стресс-тестирования: https://github.com/zznate/cassandra-stress/blob/master/src/main/java/com/riptano/cassandra/stress/InsertCommand.java

Кроме того, он немного старше, но вот суть подхода к параллельному загрузчику, который работает аналогично тому, что вы описываете: https://gist.github.com/397574