java - Cassandra 批量加载器

Question

public class SimpleLoadExample
{
    /** Default output directory */
    public static final String DEFAULT_OUTPUT_DIR = "/home/tamil/bdata";

    public static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd");

    /** Keyspace name */
    public static final String KEYSPACE = "test";
    /** Table name */
    public static final String TABLE = "counter";

    public static final String SCHEMA = "CREATE TABLE test.counter (" +
            "id varchar PRIMARY KEY," +
            "count counter" +
            ")";

    public static void main(String[] args) throws InvalidRequestException, IOException {
        Config.setClientMode(true);
        File outputDir = new File(DEFAULT_OUTPUT_DIR + File.separator + KEYSPACE + File.separator + TABLE);
        if (!outputDir.exists() && !outputDir.mkdirs())
        {
            throw new RuntimeException("Cannot create output directory: " + outputDir);
        }
        SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(outputDir, new Murmur3Partitioner(), KEYSPACE, TABLE, CounterColumnType.instance, null, 1000);
        writer.newRow(bytes("2"));
        writer.addCounterColumn(bytes("count"), 10L);
        try
        {
            writer.close();
        }
        catch (IOException ignore) {
            System.out.println(ignore);
        }
    }
}

我尝试了上面的简单示例来生成 sstables。但是

ERROR 21:05:14,287 Missing component: /home/tamil/bdata/test/counter/test-counter-jb-1-Summary.db
    [
    {"key": "32","columns": [Exception in thread "main" java.lang.IllegalArgumentException
        at java.nio.Buffer.limit(Buffer.java:267)
        at org.apache.cassandra.utils.ByteBufferUtil.readBytes(ByteBufferUtil.java:587)
        at org.apache.cassandra.utils.ByteBufferUtil.readBytesWithShortLength(ByteBufferUtil.java:596)
        at org.apache.cassandra.db.marshal.AbstractCompositeType.getString(AbstractCompositeType.java:203)
        at org.apache.cassandra.tools.SSTableExport.serializeColumn(SSTableExport.java:183)
        at org.apache.cassandra.tools.SSTableExport.serializeAtom(SSTableExport.java:152)
        at org.apache.cassandra.tools.SSTableExport.serializeAtoms(SSTableExport.java:140)
        at org.apache.cassandra.tools.SSTableExport.serializeRow(SSTableExport.java:238)
        at org.apache.cassandra.tools.SSTableExport.serializeRow(SSTableExport.java:223)
        at org.apache.cassandra.tools.SSTableExport.export(SSTableExport.java:374)
        at org.apache.cassandra.tools.SSTableExport.export(SSTableExport.java:399)
        at org.apache.cassandra.tools.SSTableExport.export(SSTableExport.java:411)
        at org.apache.cassandra.tools.SSTableExport.main(SSTableExport.java:494)

有人可以指出我哪里出错了吗？

This这是我在网上找到的。但我认为我的比较器[尝试了 UTF8Type 和 CounterColumn] 完全没问题。还有什么可能导致此缓冲区问题？

Answer 1

最后，下面的代码起作用了。发布它作为引用供人们关注

SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(outputDir, new Murmur3Partitioner(), KEYSPACE, TABLE, CompositeType.getInstance(UTF8Type.instance), null, 1000);
writer.newRow(bytes("2"));
ByteBuffer name = CompositeType.getInstance(UTF8Type.instance).builder().add(bytes("count")).build();
writer.addCounterColumn(name, 10L);

如果你能做到这一点就更好了

// SCHEMA is CQL Create table Query String  and KEYSPACE is string keyspace name
CFMetaData cfm = CFMetaData.compile(SCHEMA, KEYSPACE);
SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(outputDir, cfm, new Murmur3Partitioner(), defaultSize);