如何用Java扫描无索引的100亿条记录的Cassandra全表？

1. 基于Token范围拆分扫描 #

Cassandra的数据按Partition Token分布在节点上，我们可以手动拆分Token范围，逐个执行查询，避开单查询20亿行的限制。

实现步骤： #

• 确定Token范围：Cassandra的Token取值区间是-9223372036854775808到9223372036854775807，将其拆分成若干小范围（比如100个）。
• 循环遍历每个Token范围，执行带WHERE token(partition_key) > ? AND token(partition_key) <= ?的条件查询。
• 用fetchSize控制每次拉取的行数，避免内存溢出，逐行处理数据。

代码示例： #

@Autowired
private CassandraTemplate cassandraTemplate;

public void fullTableScan() {
    long minToken = -9223372036854775808L;
    long maxToken = 9223372036854775807L;
    // 拆分Token范围，可根据表的数据量调整拆分数量
    long step = (maxToken - minToken) / 100;

    for (long start = minToken; start < maxToken; start += step) {
        long end = Math.min(start + step, maxToken);
        // 替换为你的表名和分区键
        String cql = "SELECT * FROM abc.abc WHERE token(id) > ? AND token(id) <= ?";
        ResultSet resultSet = cassandraTemplate.getCqlOperations().queryForResultSet(
            SimpleStatement.builder(cql)
                .addPositionalValues(start, end)
                .setFetchSize(10000) // 控制单次拉取行数
                .build()
        );

        // 逐行处理数据，添加你的分组逻辑
        for (Row row : resultSet) {
            processRow(row);
        }
    }
}

private void processRow(Row row) {
    // 提取字段、执行分组统计等操作
}

2. 使用Spark Cassandra Connector #

如果允许引入Spark，Spark Cassandra Connector可以自动拆分Token范围并分布式处理，适合超大规模数据的扫描和分组。

核心步骤： #

• 引入Spark Cassandra依赖，配置连接Cassandra的参数（接触点、端口等）。
• 读取整张表，通过Spark的分布式能力进行分组处理。

代码示例： #

SparkConf conf = new SparkConf()
    .setAppName("CassandraFullScan")
    .setMaster("local[*]") // 生产环境替换为集群地址
    .set("spark.cassandra.connection.host", "your-cassandra-host")
    .set("spark.cassandra.connection.port", "9042");

JavaSparkContext sc = new JavaSparkContext(conf);

// 读取Cassandra表，指定需要的字段减少数据传输
JavaRDD<Row> cassandraRDD = javaFunctions(sc)
    .cassandraTable("abc", "abc")
    .select("col1", "col2", "col3");

// 按col1分组统计数量的示例
JavaPairRDD<String, Long> groupedRDD = cassandraRDD
    .mapToPair(row -> new Tuple2<>(row.getString("col1"), 1L))
    .reduceByKey(Long::sum);

// 处理分组结果
groupedRDD.foreach(tuple -> System.out.println(tuple._1 + ": " + tuple._2));

sc.stop();

3. 优化注意事项 #

• Token范围拆分：如果表的分区键数据分布不均，可调整拆分的范围数量，避免个别范围数据量过大。
• fetchSize调优：根据服务器配置和网络情况调整，建议取值10000-50000，平衡请求次数和内存压力。
• 限流与重试：大规模扫描会给集群带来压力，可添加限流机制或重试逻辑，避免节点过载。
• 字段过滤：仅查询需要的字段，减少数据传输量和内存占用。

内容的提问来源于stack exchange，提问作者Anish