我想将旧的Cassandra集群迁移到一个新的集群。
要求:-
我有一个包含 10 个节点的 cassandra 集群,我要迁移的表是 ~100GB。我正在使用火花来迁移数据。我的 Spark 群集有 10 个节点,每个节点有大约 16GB 的内存。在表中,我们有一些垃圾数据,我不想将其迁移到新表中。例如:- 假设我不想传输 cid = 1234 的行。那么,使用火花作业迁移此内容的最佳方法是什么?我不能直接在 cassandraRdd 上放置 where 过滤,因为 cid 不是分区键中包含的唯一列。
卡桑德拉表:-
test_table (
cid text,
uid text,
key text,
value map<text, timestamp>,
PRIMARY KEY ((cid, uid), key)
)
示例数据:-
cid | uid | key | value
------+--------------------+-----------+-------------------------------------------------------------------------
1234 | 899800070709709707 | testkey1 | {'8888': '2017-10-22 03:26:09+0000'}
6543 | 097079707970709770 | testkey2 | {'9999': '2017-10-20 11:08:45+0000', '1111': '2017-10-20 15:31:46+0000'}
我在想下面这样的事情。但我想这不是最有效的方法。
val filteredRdd = rdd.filter { row => row.getString("cid") != "1234" }
filteredRdd.saveToCassandra(KEYSPACE_NAME,NEW_TABLE_NAME)
这里最好的方法是什么?
这种方法非常好。您可能希望在数据帧中编写它以利用行编码,但这可能只有一点好处。此操作的关键瓶颈将是从Cassandra写入和读取。
DF 示例spark
.read
.format("org.apache.spark.sql.cassandra")
.option("keyspace", ks)
.option("table", table)
.load
.filter( 'cid !== "1234" )
.write
.format("org.apache.spark.sql.cassandra")
.option("keyspace", ks2)
.option("table", table2)
.save