3

这个简单的应用程序需要一个带有标题的逗号分隔文件并放入 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;
    }

}
4

3 回答 3

1

如果输入文件中的“列”之一大于分配的堆，我可以认为这是一个问题。您可以通过设置突变大小的上限来解决此问题，因为您的 CassandraSave 函数在一次操作中仅执行 100 个左右的突变。

于 2011-08-26T15:43:54.137 回答
1

我看到了两件事——它是单线程的，而且批量很小。

添加一个外部循环以收集 mutator 中的插入，批量大小约为 500 行，以启动并查看情况如何。这是我用于压力测试的高性能突变插入示例： https ://github.com/zznate/cassandra-stress/blob/master/src/main/java/com/riptano/cassandra/stress/InsertCommand.java

此外，它有点旧，但这里有一个并行加载器方法的要点，它的工作方式与您描述的类似： https ://gist.github.com/397574

于 2011-09-02T15:58:18.807 回答
1

“sun.misc.Launcher$AppClassLoader @ 0x899902f8”加载的“com.ecyrd.speed4j.log.PeriodicalLog”的一个实例占用了 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 中默认禁用，如果您想启用它，请参阅此线程。

于 2011-08-31T00:53:53.547 回答