3

我想在 cassandra 中插入大约 5000 万行（每行约 30 列），目前只有 1 个节点。

我正在从另一个数据源查询我的数据并存储在一个表对象中。我遍历单独解析每一行，然后将其添加到 mutator。目前我一次插入 100 行，100 万行需要 40 分钟！如何加快这个过程？（我也尝试过 client.batch_mutate() 但它似乎在插入几千块块大小 2 后重置了连接错误）。

通过四处搜索，我发现多线程可以提供帮助。但是我找不到任何例子，有人可以链接我吗？谢谢你 ！！

我当前的代码：

        List<String> colNames = new ArrayList<String>();
        List<String> colValues = new ArrayList<String>();
        SomeTable result = Query(...); // this contains my result set of 1M rows initially

        for (Iterator itr = result.getRecordIterator(); itr.hasNext();) {
                String colName =.....
                String colValue = .....

            int colCount = colNames.size(); // 100 * 30

            for (int i = 0; i < colCount; i++) {
                //add row keys and columns to mutator 
                mutator.addInsertion(String.valueOf(rowCounter), "data", HFactory.createStringColumn(colNames.get(i), colValues.get(i)));
            }
            rowCounter++;

            //insert rows of block size 100
            if (rowCounter % 100==0) { 

                mutator.execute();
                //clear data
                colNames = new ArrayList<String>();
                colValues = new ArrayList<String>();
                mutator = HFactory.createMutator(keyspace, stringSerializer);
            }

        }
4

1 回答 1

2

多线程会有很大帮助，是的。目前，您在 Cassandra 中使用一个连接，这意味着您在 Cassandra 中仅使用一个线程。您需要使用多个连接，这需要客户端中的多个线程。

一种方法是使用 Java ThreadPoolExecutor 并将您的 mutator.execute() 包装在一个可运行文件中并在线程池上执行它。注意处理异常。您还应该使用 BlockingQueue 来限制排队突变的数量，以防您读取源代码的速度快于 Cassandra 插入的速度。

有了这个，将 Hector 中的连接池大小设置为 10，您的插入应该会明显更快。

顺便说一句，如果您不知道，Cassandra 并不是为单节点操作而设计的。我假设您打算扩展和添加复制。如果没有，那么您可能会找到一种更高效、更简单的替代解决方案来满足您的需求。使用多个节点时，多个连接和线程变得尤为重要，因此您的插入率可以扩展。

于 2013-04-04T08:40:22.360 回答