如何使用datastax java驱动程序有效地使用批量写入cassandra？

Question

我需要使用 Datastax Java 驱动程序将批处理写入 Cassandra，这是我第一次尝试将批处理与 datastax Java 驱动程序一起使用，所以我有些困惑 -

下面是我的代码，我在其中尝试创建一个 Statement 对象并将其添加到 Batch 并将 ConsistencyLevel 设置为 QUORUM。

Session session = null;
Cluster cluster = null;

// we build cluster and session object here and we use  DowngradingConsistencyRetryPolicy as well
// cluster = builder.withSocketOptions(socketOpts).withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)

public void insertMetadata(List<AddressMetadata> listAddress) {
    // what is the purpose of unloggedBatch here?
    Batch batch = QueryBuilder.unloggedBatch();

    try {
        for (AddressMetadata data : listAddress) {
            Statement insert = insertInto("test_table").values(
                    new String[] { "address", "name", "last_modified_date", "client_id" },
                    new Object[] { data.getAddress(), data.getName(), data.getLastModifiedDate(), 1 });
            // is this the right way to set consistency level for Batch?
            insert.setConsistencyLevel(ConsistencyLevel.QUORUM);
            batch.add(insert);
        }

        // now execute the batch
        session.execute(batch);
    } catch (NoHostAvailableException e) {
        // log an exception
    } catch (QueryExecutionException e) {
        // log an exception
    } catch (QueryValidationException e) {
        // log an exception
    } catch (IllegalStateException e) {
        // log an exception
    } catch (Exception e) {
        // log an exception
    }
}

下面是我的AddressMetadata类-

public class AddressMetadata {

    private String name;
    private String address;
    private Date lastModifiedDate;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public Date getLastModifiedDate() {
        return lastModifiedDate;
    }

    public void setLastModifiedDate(Date lastModifiedDate) {
        this.lastModifiedDate = lastModifiedDate;
    }
}

现在我的问题是 - 我使用 Batch 通过 Datastax Java 驱动程序插入 cassandra 的方式是否正确？那么重试策略呢，也就是说如果批处理语句执行失败了，那会发生什么，会重试吗？

还有没有更好的方法来使用 java 驱动程序对 cassandra 进行批量写入？

Answer 1

首先有点咆哮：

Cassandra 中的批处理关键字不是用于将大型数据桶批处理以进行批量加载的性能优化。

批次用于将原子操作组合在一起，即您希望一起发生的操作。批次保证如果您的批次的单个部分成功，则整个批次都成功。

使用批处理可能不会让您的大量摄取运行得更快

现在回答您的问题：

这里 unloggedBatch 的目的是什么？

Cassandra 使用一种称为批处理日志记录的机制来确保批处理的原子性。通过指定未记录的批处理，您将关闭此功能，因此批处理不再是原子的，并且可能会因部分完成而失败。自然，记录批次并确保它们的原子性会降低性能，使用未记录的批次将消除这种损失。

在某些情况下，您可能希望使用未记录的批处理来确保将属于同一分区的请求（插入）一起发送。如果您一起批处理操作并且它们需要在不同的分区/节点中执行，那么您实际上是在为您的协调器创建更多的工作。请参阅 Ryan 的博客中的具体示例：

Read this post

现在我的问题是 - 我使用 Batch 插入的方式是否 cassandra 与 Datastax Java Driver 是否正确？

我认为您的代码没有任何问题，这取决于您要实现的目标。深入了解我分享的那篇博文以获得更多见解。

那么重试策略呢，这意味着如果批处理语句执行 失败了，那会发生什么，会重试吗？

如果批处理失败，它不会自行重试。驱动程序确实有重试策略，但您必须单独应用这些策略。

java驱动中的默认策略只在这些场景下重试：

• 在读取超时时，如果有足够多的副本回复但没有数据 已检索。
• 在写入超时时，如果我们在写入时超时 批处理语句使用的分布式日志。

阅读有关default policy 的更多信息，并根据您的用例考虑less conservative policies。

Answer 2

我们在使用异步和批处理之间争论了一段时间。我们尝试了两者进行比较。与单个“异步”请求相比，我们使用“未记录的批次”获得了更好的吞吐量。我们不知道为什么，但基于Ryan's blog，我猜这与写入大小有关。我们可能正在进行太多较小的写入，因此将它们批处理可能会给我们带来更好的性能，因为它确实减少了网络流量。

我不得不提一下，我们甚至没有按照推荐的方式进行“未记录的批次”。推荐的方法是使用单分区键进行批处理。基本上，批处理属于同一分区键的所有记录。但是，我们只是批处理了一些可能属于不同分区的记录。

有人做了一些基准测试来比较异步和“未记录的批次”，我们发现这非常有用。这是link。