如何高效拆分大型Cassandra表以进行ETL迁移？

针对你遇到的按时间分片迁移时重复全表扫描、资源浪费且无法断点续传的问题，给出以下几个实用优化策略：

1. 预收集Token范围-时间映射，缩小扫描范围 #

核心思路是先通过一次轻量扫描，获取每个Spark分区（对应Cassandra的Token范围）内交易时间的极值，后续迁移作业仅针对包含目标时间数据的Token范围进行查询，避免全表扫描：

步骤： #

• 第一步：收集Token范围与时间边界的映射
  执行一次仅查询每个分区内min(tx_time)和max(tx_time)的轻量作业，同时记录该分区对应的Token范围（Spark Cassandra Connector的每个RDD分区会携带对应的Token上下界信息）。示例代码（Java）：

  import org.apache.spark.sql.cassandra.CassandraPartition;
  import scala.Tuple2;
  import java.io.File;
  import java.io.PrintWriter;
  import java.util.Collections;
  import java.util.Map;
  
  // 读取全表但仅保留tx_time，获取每个分区的时间极值和Token范围
  Map<String, Tuple2<Long, Long>> tokenTimeMap = sparkSession.read()
          .format("org.apache.spark.sql.cassandra")
          .options(Map.of(
              "keyspace", keyspace,
              "table", table
          ))
          .select("tx_time")
          .rdd()
          .mapPartitionsWithIndex((index, iter) -> {
              // 获取当前分区对应的Cassandra Token范围
              CassandraPartition cp = (CassandraPartition) org.apache.spark.TaskContext.get().partition();
              String tokenRange = cp.getStartToken() + "-" + cp.getEndToken();
              
              // 计算当前分区的时间极值
              long minTime = Long.MAX_VALUE;
              long maxTime = Long.MIN_VALUE;
              while (iter.hasNext()) {
                  org.apache.spark.sql.Row row = iter.next();
                  java.sql.Timestamp ts = row.getTimestamp(0);
                  if (ts != null) {
                      long time = ts.getTime();
                      minTime = Math.min(minTime, time);
                      maxTime = Math.max(maxTime, time);
                  }
              }
              // 返回Token范围与时间极值的映射
              return Collections.singletonList(new Tuple2<>(tokenRange, new Tuple2<>(minTime, maxTime))).iterator();
          }, true)
          .collectAsMap();
  
  // 将映射关系保存到本地文件或轻量存储（如SQLite），供后续作业使用
  try (PrintWriter writer = new PrintWriter(new File("token_time_mapping.txt"))) {
      tokenTimeMap.forEach((tokenRange, timeRange) -> {
          writer.println(tokenRange + "," + timeRange._1() + "," + timeRange._2());
      });
  }
  

• 第二步：基于时间分片筛选目标Token范围
  后续每次迁移作业时，先读取token_time_mapping，筛选出与当前时间分片[startTime, endTime]有交集的Token范围，然后在Spark读取时指定这些Token范围，同时加上时间过滤：

  import java.util.ArrayList;
  import java.util.List;
  import org.apache.spark.sql.functions.col;
  
  // 读取预先生成的Token-Time映射，筛选符合当前时间分片的Token范围
  List<String> targetTokenRanges = new ArrayList<>();
  long startTimeMs = startTime.getTime();
  long endTimeMs = endTime.getTime();
  // 假设已从文件加载tokenTimeMap到内存
  for (Map.Entry<String, Tuple2<Long, Long>> entry : tokenTimeMap.entrySet()) {
      long minTxTime = entry.getValue()._1();
      long maxTxTime = entry.getValue()._2();
      // 判断Token范围与当前时间分片是否有交集
      if (maxTxTime >= startTimeMs && minTxTime < endTimeMs) {
          targetTokenRanges.add(entry.getKey());
      }
  }
  
  // 读取Cassandra时指定目标Token范围，结合时间过滤
  var df = sparkSession.read()
          .format("org.apache.spark.sql.cassandra")
          .options(Map.of(
              "keyspace", keyspace,
              "table", table,
              // 指定要扫描的Token范围，多个范围用逗号分隔
              "spark.cassandra.input.token.ranges", String.join(",", targetTokenRanges)
          ))
          .where(col("tx_time").geq(startTime).and(col("tx_time").lt(endTime)))
          .load();
  
  // 后续转换与写入目标数据库逻辑
  

  这样每个迁移作业只会扫描包含目标时间数据的Token分区，避免了全表扫描。

2. 实现断点续传机制 #

针对作业中断无法恢复的问题，可通过记录已完成迁移的时间分片来实现断点续传：

• 维护一个状态存储（如本地文件、Redis或小型关系数据库），记录每个已成功迁移的时间分片（如2023-01-01 00:00:00~2023-01-01 01:00:00）。
• 每次作业启动时，先读取状态存储，筛选出未完成的时间分片进行处理。
• 每个时间分片迁移完成后，将该分片标记为已完成并写入状态存储。

3. 强化过滤条件下推，减少数据传输 #

确保Spark Cassandra Connector将时间过滤条件推送到Cassandra端执行，避免在Spark端加载大量不符合条件的数据：

• 确认开启过滤下推（默认开启，可显式设置）：

  sparkSession.conf().set("spark.sql.cassandra.filterPushdown", "true");
  

• 使用where()而非filter()方法添加时间条件（两者在Spark SQL中语义一致，但where()更利于Connector识别并下推谓词）。
• 避免在load()之后再进行复杂的转换操作，尽量先过滤再处理，减少数据在集群中的传输量。

4. 按Token范围拆分+时间过滤的组合方案 #

你之前尝试的按Token范围拆分并非不可行，Spark Cassandra Connector支持按Token范围拆分作业，可结合时间过滤实现更细粒度的并行处理：

• 获取Cassandra集群的所有Token范围（可通过nodetool ring命令或Cassandra Java API获取）。
• 将Token范围拆分为若干批次，每个批次处理部分Token范围，同时在每个批次中添加时间过滤条件。
• 结合断点续传机制，记录已处理的Token范围和时间分片，避免重复处理。

内容的提问来源于stack exchange，提问作者つきがきれいだね