如何在Apache Cassandra中导出近30天或限定数据并迁移至其他实例

针对你需要筛选近30天数据导出并迁移的需求，以下是几种实用的实现方法：

方法一：CQL查询筛选 + COPY命令（简单易操作） #

CQLSH的COPY命令支持直接导出查询结果，你可以先通过带时间条件的CQL语句过滤出近30天的数据，再导出到CSV，最后导入目标集群：

1. 源集群导出数据：
   在cqlsh中执行以下命令（替换your_keyspace、your_table和timestamp_column为实际值）：

   COPY (SELECT * FROM your_keyspace.your_table WHERE timestamp_column >= toTimestamp(now()) - 30d) TO '/local/path/30days_backup.csv' WITH HEADER = TRUE;
   

   • 30d是Cassandra原生支持的时间间隔语法，也可以用毫秒数替代（比如2592000000，对应30天的毫秒数）；
   • 如果表中有集合、UDT等复杂类型，需额外配置DELIMITER、QUOTE等参数适配数据格式。

2. 目标集群导入数据：
   将CSV文件传输到目标节点后，执行导入命令：

   COPY your_keyspace.your_table FROM '/local/path/30days_backup.csv' WITH HEADER = TRUE;
   

方法二：sstable2json + json2sstable（适合批量高性能场景） #

如果数据量较大，直接用CQL查询效率不高，可以利用Cassandra的SSTable工具链来处理：

1. 导出SSTable为JSON：
   先找到源表对应的SSTable存储路径（默认路径为/var/lib/cassandra/data/your_keyspace/your_table-<UUID>/），然后用sstable2json导出：

   sstable2json /var/lib/cassandra/data/your_keyspace/your_table-<UUID>/mc-1-big-Data.db > raw_data.json
   

2. 过滤JSON中的近30天数据：
   用脚本工具（如jq或Python）筛选符合时间条件的记录，比如用jq：

   jq 'select(.timestamp_column >= (now - 30*86400)*1000)' raw_data.json > filtered_backup.json
   

3. 转换为SSTable并加载到目标集群：
   将过滤后的JSON传到目标节点，用json2sstable转换为SSTable，再通过nodetool加载：

   json2sstable -k your_keyspace -t your_table filtered_backup.json /tmp/temp_sstable
   nodetool refresh your_keyspace your_table
   

   注意：操作前需确保目标节点处于正常状态，避免数据不一致。

方法三：Apache Spark（适合大规模复杂迁移） #

如果数据量极大或需要复杂的转换逻辑，用Spark连接源和目标Cassandra集群是更高效的选择：

示例Scala代码片段（需依赖Spark Cassandra Connector）：

import org.apache.spark.sql.cassandra._
import java.time.Instant

// 计算30天前的时间戳（毫秒）
val thirtyDaysAgo = Instant.now().minusSeconds(30 * 86400).toEpochMilli

// 读取源集群中近30天的数据
val sourceDF = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "your_table", "keyspace" -> "your_keyspace"))
  .load()
  .filter(s"timestamp_column >= $thirtyDaysAgo")

// 写入目标集群
sourceDF.write
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "your_table", "keyspace" -> "your_keyspace"))
  .mode("append")
  .save()

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