如何从 Cassandra 流式传输所有记录？答案

【问题标题】：How to stream all records from Cassandra?如何从 Cassandra 流式传输所有记录？
【发布时间】：2021-12-04 18:30:20
【问题描述】：

我需要从 Cassandra 流式传输所有记录。目前我正在使用akka-persistence-cassandra 流式传输数据：

val querier =
        PersistenceQuery(system)
          .readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
      
      val selectDistinctPersistenceIds = new SimpleStatement(
      "SELECT DISTINCT persistence_id, partition_nr FROM messages")
        .setFetchSize(100000)

        querier.session.select(selectDistinctPersistenceIds).map { row =>
          val id = row.getString(0)
          id
        }

当记录数约为 150 万时，此方法可以正常工作。但是当记录数超过> 150万条记录时，我得到read timeout错误。

我正在使用：

"com.typesafe.akka" %% "akka-persistence-cassandra" % "0.58"
"com.typesafe.akka" %% "akka-persistence" % "2.6.12"
"com.typesafe.akka" %% "akka-persistence-query" % "2.6.12"

编辑：错误日志：

com.datastax.driver.core.exceptions.OperationTimedOutException: [/<ip-address>:9042] Timed out waiting for server response", exceptionStackTrace="java.util.concurrent.ExecutionException: com.datastax.driver.core.exceptions.OperationTimedOutException: [/<ip-address>:9042] Timed out waiting for server response
    at com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:552)
    at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:513)
    at akka.persistence.cassandra.package$ListenableFutureConverter$$anon$2.$anonfun$run$2(package.scala:25)
...

【问题讨论】：

标签： scala cassandra akka

【解决方案1】：

问题在于您的驱动程序会话设置根据您的需要进行调整。

gap-timeout 可能有问题或增加重试次数。和超时设置。

【讨论】：

很遗憾，我无法更改任何设置，因为我不确定 Cassandra 中有多少数据。
您有日志吗？错误是什么，它将提供一些线索，否则无法调试问题
“我不确定 Cassandra 中有多少数据。”在尝试将其全部流式传输之前，您确实应该知道这一点。 @himanshuIIITian
对于大约 800 万条记录的估计，上述代码失败。因此数据是 >8M 记录。
这里清楚地告诉 OperationTimedOutException 所以你必须增加超时设置

【解决方案2】：

我可以通过为cassandra-journal.socket.read-timeout-millis 设置一个高于默认值 12000 毫秒的值来解决此问题。

cassandra-journal {
  ...

  socket {
    # the per-host read timeout in milliseconds. Should be higher than the timeout settings
    # used on the Cassandra side.
    read-timeout-millis = 30000
}

【讨论】：