Cassandra 计数查询抛出 ReadFailureException答案

【问题标题】：Cassandra count query throwing ReadFailureExceptionCassandra 计数查询抛出 ReadFailureException
【发布时间】：2020-01-18 13:08:32
【问题描述】：

我正在使用 spark-sql 2.4.1、spark-cassandra-connector_2.11-2.4.1.jar 和 java8。我有一种情况，出于审计目的，我需要计算 C* 表的表行数。我的 C* 表中有大约 20 亿条记录。

为了计算行数，我尝试了两种方法，如下所示。

public static Long getColumnFamilyCountJavaApi(SparkSession spark,String keyspace, String columnFamilyName)  throws IOException{
  JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
  return javaFunctions(sc).cassandraTable(keyspace, columnFamilyName).cassandraCount();
}

public static Long getColumnFamilyCount(SparkSession spark,String keyspace, String columnFamilyName)  throws IOException{
  return spark
              .read()
              .format("org.apache.spark.sql.cassandra")
              .option("table", columnFamilyName)
              .option("keyspace",keyspace )
              .load().count();
}

但两种方式都会导致相同的错误。

   Caused by: com.datastax.driver.core.exceptions.ReadFailureException: Cassandra failure during read query at consistency LOCAL_QUORUM (2 responses were required but only 0 replica responded, 2 failed)
            at com.datastax.driver.core.exceptions.ReadFailureException.copy(ReadFailureException.java:85)
    com.datastax.driver.core.DefaultResultSetFuture.getUninterruptibly(DefaultResultSetFuture.java:245)
            at com.datastax.spark.connector.cql.DefaultScanner.scan(Scanner.scala:34)
            at com.datastax.spark.connector.rdd.CassandraTableScanRDD.com$datastax$spark$connector$rdd$CassandraTableScanRDD$$fetchTokenRange(CassandraTableScanRDD.scala:342)

如何处理这种情况？

【问题讨论】：

标签： apache-spark cassandra apache-spark-sql datastax-enterprise datastax-java-driver

【解决方案1】：

该错误堆栈是节点的读取超时。这实际上可能是由于多种原因。而不是回答这个特定的错误，我将在你的最终目标的背景下回答。

您正在尝试在 Cassandra 中计算表中的行数。

虽然这不是一个不合理的要求，但对于 Cassandra 来说，这是一个有点棘手的话题。这是因为计数是集群范围的。看看这个相当不错的blog article，了解为什么会这样。

我可以看到您在此处使用 spark，因此您可能已经意识到在 CQLSH 中计数可能会很昂贵。你可能想看看academy video here 的cassandraCount 也可以看看spark connector docs

您可能还对 DSbulk 工具感兴趣。我已经成功地将这个工具用于从大型数据迁移到计数等小型作业的许多事情。请参阅DSbulk docs here

希望对大家有所帮助！

【讨论】：